diff --git a/dd-trace-core/src/jmh/java/datadog/trace/core/monitor/LegacyTracerHealthMetrics.java b/dd-trace-core/src/jmh/java/datadog/trace/core/monitor/LegacyTracerHealthMetrics.java
new file mode 100644
index 00000000000..0a08f2b6d90
--- /dev/null
+++ b/dd-trace-core/src/jmh/java/datadog/trace/core/monitor/LegacyTracerHealthMetrics.java
@@ -0,0 +1,264 @@
+package datadog.trace.core.monitor;
+
+import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_DROP;
+import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_KEEP;
+import static datadog.trace.api.sampling.PrioritySampling.USER_DROP;
+import static datadog.trace.api.sampling.PrioritySampling.USER_KEEP;
+
+import datadog.metrics.api.statsd.StatsDClient;
+import datadog.trace.common.writer.RemoteApi;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * A faithful reconstruction of the pre-{@code Accumulator} {@code TracerHealthMetrics} -- one
+ * {@link LongAdder} field per counter, hand-rolled switch statements, a hand-concatenated {@code
+ * summary()} -- as it stood at {@code 77964b3996} (the last commit on {@code master} before the
+ * migration), restricted to the exact methods {@link TracerHealthMetricsBenchmark} exercises. Kept
+ * as a standalone class here (not resurrected via checkout) purely for a same-run, same-JVM
+ * before/after comparison; it is not wired into anything and should never be.
+ */
+class LegacyTracerHealthMetrics {
+
+ private final LongAdder apiRequests = new LongAdder();
+ private final LongAdder apiErrors = new LongAdder();
+ private final LongAdder apiResponsesOK = new LongAdder();
+
+ private final LongAdder userDropEnqueuedTraces = new LongAdder();
+ private final LongAdder userKeepEnqueuedTraces = new LongAdder();
+ private final LongAdder samplerDropEnqueuedTraces = new LongAdder();
+ private final LongAdder samplerKeepEnqueuedTraces = new LongAdder();
+ private final LongAdder unsetPriorityEnqueuedTraces = new LongAdder();
+
+ private final LongAdder userDropDroppedTraces = new LongAdder();
+ private final LongAdder userKeepDroppedTraces = new LongAdder();
+ private final LongAdder samplerDropDroppedTraces = new LongAdder();
+ private final LongAdder samplerKeepDroppedTraces = new LongAdder();
+ private final LongAdder serialFailedDroppedTraces = new LongAdder();
+ private final LongAdder unsetPriorityDroppedTraces = new LongAdder();
+
+ private final LongAdder userDropDroppedSpans = new LongAdder();
+ private final LongAdder userKeepDroppedSpans = new LongAdder();
+ private final LongAdder samplerDropDroppedSpans = new LongAdder();
+ private final LongAdder samplerKeepDroppedSpans = new LongAdder();
+ private final LongAdder serialFailedDroppedSpans = new LongAdder();
+ private final LongAdder unsetPriorityDroppedSpans = new LongAdder();
+
+ private final LongAdder enqueuedSpans = new LongAdder();
+ private final LongAdder enqueuedBytes = new LongAdder();
+ private final LongAdder createdTraces = new LongAdder();
+ private final LongAdder createdSpans = new LongAdder();
+ private final LongAdder finishedSpans = new LongAdder();
+ private final LongAdder flushedTraces = new LongAdder();
+ private final LongAdder flushedBytes = new LongAdder();
+ private final LongAdder partialTraces = new LongAdder();
+ private final LongAdder partialBytes = new LongAdder();
+ private final LongAdder clientSpansWithoutContext = new LongAdder();
+
+ private final LongAdder singleSpanSampled = new LongAdder();
+ private final LongAdder singleSpanUnsampled = new LongAdder();
+
+ private final LongAdder capturedContinuations = new LongAdder();
+ private final LongAdder cancelledContinuations = new LongAdder();
+ private final LongAdder finishedContinuations = new LongAdder();
+
+ private final LongAdder activatedScopes = new LongAdder();
+ private final LongAdder closedScopes = new LongAdder();
+ private final LongAdder scopeStackOverflow = new LongAdder();
+ private final LongAdder scopeCloseErrors = new LongAdder();
+ private final LongAdder userScopeCloseErrors = new LongAdder();
+
+ private final LongAdder longRunningTracesWrite = new LongAdder();
+ private final LongAdder longRunningTracesDropped = new LongAdder();
+ private final LongAdder longRunningTracesExpired = new LongAdder();
+
+ private final LongAdder clientStatsProcessedSpans = new LongAdder();
+ private final LongAdder clientStatsProcessedTraces = new LongAdder();
+ private final LongAdder clientStatsP0DroppedSpans = new LongAdder();
+ private final LongAdder clientStatsP0DroppedTraces = new LongAdder();
+ private final LongAdder clientStatsRequests = new LongAdder();
+ private final LongAdder clientStatsErrors = new LongAdder();
+ private final LongAdder clientStatsDowngrades = new LongAdder();
+
+ private final LongAdder statsAggregateDropped = new LongAdder();
+ private final LongAdder statsInboxFull = new LongAdder();
+
+ private final StatsDClient statsd;
+
+ LegacyTracerHealthMetrics(StatsDClient statsd) {
+ this.statsd = statsd;
+ }
+
+ void onCreateSpan() {
+ createdSpans.increment();
+ }
+
+ void onFailedPublish(final int samplingPriority, final int spanCount) {
+ switch (samplingPriority) {
+ case USER_DROP:
+ userDropDroppedSpans.add(spanCount);
+ userDropDroppedTraces.increment();
+ break;
+ case USER_KEEP:
+ userKeepDroppedSpans.add(spanCount);
+ userKeepDroppedTraces.increment();
+ break;
+ case SAMPLER_DROP:
+ samplerDropDroppedSpans.add(spanCount);
+ samplerDropDroppedTraces.increment();
+ break;
+ case SAMPLER_KEEP:
+ samplerKeepDroppedSpans.add(spanCount);
+ samplerKeepDroppedTraces.increment();
+ break;
+ default:
+ unsetPriorityDroppedSpans.add(spanCount);
+ unsetPriorityDroppedTraces.increment();
+ }
+ }
+
+ void onPartialPublish(final int numberOfDroppedSpans) {
+ partialTraces.increment();
+ samplerDropDroppedSpans.add(numberOfDroppedSpans);
+ }
+
+ void onSend(final int traceCount, final int sizeInBytes, final RemoteApi.Response response) {
+ onSendAttempt(traceCount, sizeInBytes, response);
+ }
+
+ private void onSendAttempt(
+ final int traceCount, final int sizeInBytes, final RemoteApi.Response response) {
+ apiRequests.increment();
+ flushedTraces.add(traceCount);
+ flushedBytes.add(sizeInBytes);
+
+ if (response.exception().isPresent()) {
+ apiErrors.increment();
+ }
+
+ int status = response.status().orElse(0);
+ if (status != 0) {
+ if (200 == status) {
+ apiResponsesOK.increment();
+ } else {
+ statsd.incrementCounter("api.responses.total", "status:" + status);
+ }
+ }
+ }
+
+ String summary() {
+ return "apiRequests="
+ + apiRequests.sum()
+ + "\napiErrors="
+ + apiErrors.sum()
+ + "\napiResponsesOK="
+ + apiResponsesOK.sum()
+ + "\n"
+ + "\nuserDropEnqueuedTraces="
+ + userDropEnqueuedTraces.sum()
+ + "\nuserKeepEnqueuedTraces="
+ + userKeepEnqueuedTraces.sum()
+ + "\nsamplerDropEnqueuedTraces="
+ + samplerDropEnqueuedTraces.sum()
+ + "\nsamplerKeepEnqueuedTraces="
+ + samplerKeepEnqueuedTraces.sum()
+ + "\nunsetPriorityEnqueuedTraces="
+ + unsetPriorityEnqueuedTraces.sum()
+ + "\n"
+ + "\nuserDropDroppedTraces="
+ + userDropDroppedTraces.sum()
+ + "\nuserKeepDroppedTraces="
+ + userKeepDroppedTraces.sum()
+ + "\nsamplerDropDroppedTraces="
+ + samplerDropDroppedTraces.sum()
+ + "\nsamplerKeepDroppedTraces="
+ + samplerKeepDroppedTraces.sum()
+ + "\nserialFailedDroppedTraces="
+ + serialFailedDroppedTraces.sum()
+ + "\nunsetPriorityDroppedTraces="
+ + unsetPriorityDroppedTraces.sum()
+ + "\n"
+ + "\nuserDropDroppedSpans="
+ + userDropDroppedSpans.sum()
+ + "\nuserKeepDroppedSpans="
+ + userKeepDroppedSpans.sum()
+ + "\nsamplerDropDroppedSpans="
+ + samplerDropDroppedSpans.sum()
+ + "\nsamplerKeepDroppedSpans="
+ + samplerKeepDroppedSpans.sum()
+ + "\nserialFailedDroppedSpans="
+ + serialFailedDroppedSpans.sum()
+ + "\nunsetPriorityDroppedSpans="
+ + unsetPriorityDroppedSpans.sum()
+ + "\n"
+ + "\nenqueuedSpans="
+ + enqueuedSpans.sum()
+ + "\nenqueuedBytes="
+ + enqueuedBytes.sum()
+ + "\ncreatedTraces="
+ + createdTraces.sum()
+ + "\ncreatedSpans="
+ + createdSpans.sum()
+ + "\nfinishedSpans="
+ + finishedSpans.sum()
+ + "\nflushedTraces="
+ + flushedTraces.sum()
+ + "\nflushedBytes="
+ + flushedBytes.sum()
+ + "\npartialTraces="
+ + partialTraces.sum()
+ + "\npartialBytes="
+ + partialBytes.sum()
+ + "\n"
+ + "\nclientSpansWithoutContext="
+ + clientSpansWithoutContext.sum()
+ + "\n"
+ + "\nsingleSpanSampled="
+ + singleSpanSampled.sum()
+ + "\nsingleSpanUnsampled="
+ + singleSpanUnsampled.sum()
+ + "\n"
+ + "\ncapturedContinuations="
+ + capturedContinuations.sum()
+ + "\ncancelledContinuations="
+ + cancelledContinuations.sum()
+ + "\nfinishedContinuations="
+ + finishedContinuations.sum()
+ + "\n"
+ + "\nactivatedScopes="
+ + activatedScopes.sum()
+ + "\nclosedScopes="
+ + closedScopes.sum()
+ + "\nscopeStackOverflow="
+ + scopeStackOverflow.sum()
+ + "\nscopeCloseErrors="
+ + scopeCloseErrors.sum()
+ + "\nuserScopeCloseErrors="
+ + userScopeCloseErrors.sum()
+ + "\n"
+ + "\nlongRunningTracesWrite="
+ + longRunningTracesWrite.sum()
+ + "\nlongRunningTracesDropped="
+ + longRunningTracesDropped.sum()
+ + "\nlongRunningTracesExpired="
+ + longRunningTracesExpired.sum()
+ + "\n"
+ + "\nclientStatsRequests="
+ + clientStatsRequests.sum()
+ + "\nclientStatsErrors="
+ + clientStatsErrors.sum()
+ + "\nclientStatsDowngrades="
+ + clientStatsDowngrades.sum()
+ + "\nclientStatsP0DroppedSpans="
+ + clientStatsP0DroppedSpans.sum()
+ + "\nclientStatsP0DroppedTraces="
+ + clientStatsP0DroppedTraces.sum()
+ + "\nclientStatsProcessedSpans="
+ + clientStatsProcessedSpans.sum()
+ + "\nclientStatsProcessedTraces="
+ + clientStatsProcessedTraces.sum()
+ + "\nstatsAggregateDropped="
+ + statsAggregateDropped.sum()
+ + "\nstatsInboxFull="
+ + statsInboxFull.sum();
+ }
+}
diff --git a/dd-trace-core/src/jmh/java/datadog/trace/core/monitor/TracerHealthMetricsBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/core/monitor/TracerHealthMetricsBenchmark.java
new file mode 100644
index 00000000000..89de8beaef5
--- /dev/null
+++ b/dd-trace-core/src/jmh/java/datadog/trace/core/monitor/TracerHealthMetricsBenchmark.java
@@ -0,0 +1,243 @@
+package datadog.trace.core.monitor;
+
+import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_DROP;
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+
+import datadog.metrics.api.statsd.StatsDClient;
+import datadog.trace.common.writer.RemoteApi;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.GroupThreads;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Direct measurement of the real {@link TracerHealthMetrics} entry points hit on the tracing hot
+ * path -- the {@link datadog.trace.util.Accumulator}-backed implementation, not a synthetic
+ * stand-in -- so the {@code AccumulatorBenchmark} numbers (raw {@code Accumulator} vs {@code
+ * LongAdder}) can be checked against what the real per-call-site shapes cost once real switches,
+ * multi-counter updates, and response-as-context dispatch are involved. All calls go through {@link
+ * StatsDClient#NO_OP} so only the accumulator-side cost is measured, not statsd transport.
+ *
+ * {@code onCreateSpan}/{@code onFinishSpan}/{@code onActivateScope}/{@code onCloseScope} are the
+ * highest-frequency calls (once per span/scope) and are each a single {@code inc()}. {@code
+ * onFailedPublish} exercises a switch plus two independent (ungrouped) counter updates. {@code
+ * onPartialPublish} exercises the boxing-free {@code update(long, ObjLongConsumer)} grouped-update
+ * path. {@code onSend} exercises {@code onSendAttempt}'s split shape: three top-level calls plus
+ * one {@code update(response, ...)} grouping the two response-derived counters under one lock.
+ *
+ *
{@code summaryWhileWriting} pairs concurrent {@code onCreateSpan} writers against a single
+ * reader repeatedly calling {@code summary()} -- {@code summary()} only peeks ({@code
+ * Accumulator#sum()}), never drains, so it should not stall the periodic {@code Flush} task nor
+ * meaningfully slow down concurrent writers; this checks that design assumption under load rather
+ * than just asserting it.
+ *
+ *
Before/after. The {@code legacy*} benchmarks run the identical inputs against {@link
+ * LegacyTracerHealthMetrics}, a faithful reconstruction of pre-migration {@code
+ * TracerHealthMetrics} (one {@code LongAdder} field per counter, hand-rolled switches, a
+ * hand-concatenated {@code summary()}) as it stood at {@code 77964b3996}, the last commit before
+ * this migration -- a same-run, same-JVM before/after comparison of the real class, not just the
+ * underlying primitive.
+ *
+ *
Results. After {@link datadog.trace.util.Accumulator}'s lock-free {@code
+ * AtomicLongArray}-striping rewrite, every hot single-counter call (uncontended) lands at
+ * 0.007-0.009 us/op -- indistinguishable from {@code AccumulatorBenchmark}'s raw {@code
+ * accumulatorIncrement_lowContention} (0.007 us/op). At {@code Threads.MAX} the real call sites
+ * stay just as flat (0.009-0.013 us/op): the CAS-based {@code getAndAdd} no longer pays the
+ * 3-4x {@code synchronized}-stripe contention penalty the earlier design did. {@code
+ * summaryWhileWriting} confirms the peek-not-drain design for {@code summary()}: concurrent
+ * readers don't measurably slow writers (0.008 us/op, same as uncontended {@code onCreateSpan}),
+ * at the cost of the read itself walking all 54 stripes non-destructively (~1.83 us/op) --
+ * acceptable for a diagnostic/tracer-flare call, never on the span-emission path. Results are
+ * stable across JDK 17 and JDK 25 (point estimates agree to the millisecond-precision printed
+ * below), confirming this is the striping rewrite's effect, not a JIT/JVM-version artifact.
+ *
+ * Apple M1 Max, 10 CPUs - macOS/aarch64 - JDK 17 (Zulu) / JDK 25 (Zulu)
+ * Benchmark JDK17 JDK25 Units
+ * TracerHealthMetricsBenchmark.onCreateSpan_lowContention 0.007 0.007 us/op
+ * TracerHealthMetricsBenchmark.onCreateSpan_highContention 0.009 0.009 us/op
+ * TracerHealthMetricsBenchmark.onFailedPublish_lowContention 0.007 0.007 us/op
+ * TracerHealthMetricsBenchmark.onFailedPublish_highContention 0.010 0.010 us/op
+ * TracerHealthMetricsBenchmark.onPartialPublish_lowContention 0.007 0.007 us/op
+ * TracerHealthMetricsBenchmark.onPartialPublish_highContention 0.009 0.010 us/op
+ * TracerHealthMetricsBenchmark.onSend_lowContention 0.009 0.009 us/op
+ * TracerHealthMetricsBenchmark.onSend_highContention 0.013 0.013 us/op
+ * TracerHealthMetricsBenchmark.summaryWhileWriting 0.373 0.373 us/op
+ * TracerHealthMetricsBenchmark.summaryWhileWriting:...write 0.008 0.008 us/op
+ * TracerHealthMetricsBenchmark.summaryWhileWriting:...read 1.835 1.833 us/op
+ *
+ *
+ *
Before/after results. The {@code Accumulator}-backed implementation is now at parity
+ * with, or measurably faster than, the {@code LongAdder} baseline it replaced on every
+ * single-call-site benchmark, including under {@code Threads.MAX} contention -- a reversal of the
+ * earlier {@code synchronized}-stripe design's 1.1-4.6x cost documented before the lock-free
+ * rewrite (commit {@code 3ea84793d1}). {@code onSend}, the most expensive real call site (three
+ * top-level calls plus one two-counter grouped update), now costs 0.6-0.75x of legacy at both
+ * contention levels. {@code summary()} remains the one real cost: walking 54 stripes
+ * non-destructively still costs ~2.5-2.6x what summing 52 plain {@code LongAdder} fields does
+ * (~1.83 vs ~0.71 us/op) -- still far below the periodic (30s-default) {@code Flush} cadence and
+ * the ad hoc/diagnostic calls that trigger it, so not disqualifying. Neither implementation's
+ * writers are measurably slowed by a concurrent {@code summary()}/reader.
+ * Apple M1 Max, 10 CPUs - macOS/aarch64 - JDK 17 (Zulu) / JDK 25 (Zulu)
+ * Benchmark New (JDK17/25) Legacy (JDK17/25) Ratio
+ * onCreateSpan_lowContention 0.007 / 0.007 0.007 / 0.007 1.0x / 1.0x
+ * onCreateSpan_highContention 0.009 / 0.009 0.010 / 0.010 0.9x / 0.9x
+ * onFailedPublish_lowContention 0.007 / 0.007 0.008 / 0.008 0.9x / 0.9x
+ * onFailedPublish_highContention 0.010 / 0.010 0.011 / 0.012 0.9x / 0.8x
+ * onPartialPublish_lowContention 0.007 / 0.007 0.008 / 0.008 0.9x / 0.9x
+ * onPartialPublish_highContention 0.009 / 0.010 0.012 / 0.012 0.75x / 0.8x
+ * onSend_lowContention 0.009 / 0.009 0.012 / 0.013 0.75x / 0.7x
+ * onSend_highContention 0.013 / 0.013 0.020 / 0.022 0.65x / 0.6x
+ * summaryWhileWriting_write 0.008 / 0.008 0.009 / 0.008 0.9x / 1.0x
+ * summaryWhileWriting_read 1.835 / 1.833 0.705 / 0.720 2.6x / 2.55x
+ * (all figures us/op, avgt; JDK17 / JDK25)
+ * This means the migration's case no longer rests solely on eliminating the {@code
+ * previousCounts}/{@code countIndex} hand-tracking ceremony and giving each counter an atomic
+ * multi-field grouped update -- the lock-free rewrite makes it a speedup too, on every path except
+ * the diagnostic {@code summary()} read.
+ */
+@State(Scope.Benchmark)
+@Warmup(iterations = 1, time = 10)
+@Measurement(iterations = 3, time = 10)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(MICROSECONDS)
+@Fork(2)
+public class TracerHealthMetricsBenchmark {
+
+ private final TracerHealthMetrics metrics = new TracerHealthMetrics(StatsDClient.NO_OP);
+ private final LegacyTracerHealthMetrics legacyMetrics =
+ new LegacyTracerHealthMetrics(StatsDClient.NO_OP);
+ private final RemoteApi.Response okResponse = RemoteApi.Response.success(200);
+
+ @Benchmark
+ @Threads(1)
+ public void onCreateSpan_lowContention() {
+ metrics.onCreateSpan();
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void onCreateSpan_highContention() {
+ metrics.onCreateSpan();
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void onFailedPublish_lowContention() {
+ metrics.onFailedPublish(SAMPLER_DROP, 5);
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void onFailedPublish_highContention() {
+ metrics.onFailedPublish(SAMPLER_DROP, 5);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void onPartialPublish_lowContention() {
+ metrics.onPartialPublish(3);
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void onPartialPublish_highContention() {
+ metrics.onPartialPublish(3);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void onSend_lowContention() {
+ metrics.onSend(1, 512, okResponse);
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void onSend_highContention() {
+ metrics.onSend(1, 512, okResponse);
+ }
+
+ @Benchmark
+ @Group("summaryWhileWriting")
+ @GroupThreads(4)
+ public void summaryWhileWriting_write() {
+ metrics.onCreateSpan();
+ }
+
+ @Benchmark
+ @Group("summaryWhileWriting")
+ @GroupThreads(1)
+ public void summaryWhileWriting_read(Blackhole blackhole) {
+ blackhole.consume(metrics.summary());
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void legacyOnCreateSpan_lowContention() {
+ legacyMetrics.onCreateSpan();
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void legacyOnCreateSpan_highContention() {
+ legacyMetrics.onCreateSpan();
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void legacyOnFailedPublish_lowContention() {
+ legacyMetrics.onFailedPublish(SAMPLER_DROP, 5);
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void legacyOnFailedPublish_highContention() {
+ legacyMetrics.onFailedPublish(SAMPLER_DROP, 5);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void legacyOnPartialPublish_lowContention() {
+ legacyMetrics.onPartialPublish(3);
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void legacyOnPartialPublish_highContention() {
+ legacyMetrics.onPartialPublish(3);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void legacyOnSend_lowContention() {
+ legacyMetrics.onSend(1, 512, okResponse);
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void legacyOnSend_highContention() {
+ legacyMetrics.onSend(1, 512, okResponse);
+ }
+
+ @Benchmark
+ @Group("legacySummaryWhileWriting")
+ @GroupThreads(4)
+ public void legacySummaryWhileWriting_write() {
+ legacyMetrics.onCreateSpan();
+ }
+
+ @Benchmark
+ @Group("legacySummaryWhileWriting")
+ @GroupThreads(1)
+ public void legacySummaryWhileWriting_read(Blackhole blackhole) {
+ blackhole.consume(legacyMetrics.summary());
+ }
+}
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/monitor/TracerHealthMetrics.java b/dd-trace-core/src/main/java/datadog/trace/core/monitor/TracerHealthMetrics.java
index f9c76ff0766..9418d7d0fe4 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/monitor/TracerHealthMetrics.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/monitor/TracerHealthMetrics.java
@@ -10,100 +10,34 @@
import static java.util.concurrent.TimeUnit.SECONDS;
import datadog.metrics.api.statsd.StatsDClient;
+import datadog.metrics.api.statsd.StatsDCountReporter;
+import datadog.metrics.api.statsd.StatsDCounterKey;
import datadog.trace.api.cache.RadixTreeCache;
import datadog.trace.common.writer.RemoteApi;
import datadog.trace.core.DDSpan;
import datadog.trace.core.propagation.opg.OrgGuard;
+import datadog.trace.util.Accumulator;
import datadog.trace.util.AgentTaskScheduler;
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.LongAdder;
import java.util.function.IntFunction;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class TracerHealthMetrics extends HealthMetrics implements AutoCloseable {
- private static final Logger log = LoggerFactory.getLogger(TracerHealthMetrics.class);
private static final IntFunction STATUS_TAGS =
httpStatus -> new String[] {"status:" + httpStatus};
private static final String[] NO_TAGS = new String[0];
private static final String[] COLLAPSED_WHOLE_KEY_TAGS = new String[] {"collapsed:whole_key"};
- private static final String[] STATUS_OK_TAGS = STATUS_TAGS.apply(200);
private final RadixTreeCache statusTagsCache =
new RadixTreeCache<>(16, 32, STATUS_TAGS, 200, 400);
private final AtomicBoolean started = new AtomicBoolean(false);
private volatile AgentTaskScheduler.Scheduled cancellation;
- private final LongAdder apiRequests = new LongAdder();
- private final LongAdder apiErrors = new LongAdder();
- private final LongAdder apiResponsesOK = new LongAdder();
-
- private final LongAdder userDropEnqueuedTraces = new LongAdder();
- private final LongAdder userKeepEnqueuedTraces = new LongAdder();
- private final LongAdder samplerDropEnqueuedTraces = new LongAdder();
- private final LongAdder samplerKeepEnqueuedTraces = new LongAdder();
- private final LongAdder unsetPriorityEnqueuedTraces = new LongAdder();
-
- private final LongAdder userDropDroppedTraces = new LongAdder();
- private final LongAdder userKeepDroppedTraces = new LongAdder();
- private final LongAdder samplerDropDroppedTraces = new LongAdder();
- private final LongAdder samplerKeepDroppedTraces = new LongAdder();
- private final LongAdder serialFailedDroppedTraces = new LongAdder();
- private final LongAdder unsetPriorityDroppedTraces = new LongAdder();
-
- private final LongAdder userDropDroppedSpans = new LongAdder();
- private final LongAdder userKeepDroppedSpans = new LongAdder();
- private final LongAdder samplerDropDroppedSpans = new LongAdder();
- private final LongAdder samplerKeepDroppedSpans = new LongAdder();
- private final LongAdder serialFailedDroppedSpans = new LongAdder();
- private final LongAdder unsetPriorityDroppedSpans = new LongAdder();
-
- private final LongAdder enqueuedSpans = new LongAdder();
- private final LongAdder enqueuedBytes = new LongAdder();
- private final LongAdder createdTraces = new LongAdder();
- private final LongAdder createdSpans = new LongAdder();
- private final LongAdder finishedSpans = new LongAdder();
- private final LongAdder flushedTraces = new LongAdder();
- private final LongAdder flushedBytes = new LongAdder();
- private final LongAdder partialTraces = new LongAdder();
- private final LongAdder partialBytes = new LongAdder();
- private final LongAdder clientSpansWithoutContext = new LongAdder();
-
- private final LongAdder singleSpanSampled = new LongAdder();
- private final LongAdder singleSpanUnsampled = new LongAdder();
-
- private final LongAdder capturedContinuations = new LongAdder();
- private final LongAdder cancelledContinuations = new LongAdder();
- private final LongAdder finishedContinuations = new LongAdder();
-
- private final LongAdder activatedScopes = new LongAdder();
- private final LongAdder closedScopes = new LongAdder();
- private final LongAdder scopeStackOverflow = new LongAdder();
- private final LongAdder scopeCloseErrors = new LongAdder();
- private final LongAdder userScopeCloseErrors = new LongAdder();
-
- private final LongAdder longRunningTracesWrite = new LongAdder();
- private final LongAdder longRunningTracesDropped = new LongAdder();
- private final LongAdder longRunningTracesExpired = new LongAdder();
-
- private final LongAdder orgGuardEnforceMismatch = new LongAdder();
- private final LongAdder orgGuardEnforceStrictMissing = new LongAdder();
-
- private final LongAdder clientStatsProcessedSpans = new LongAdder();
- private final LongAdder clientStatsProcessedTraces = new LongAdder();
- private final LongAdder clientStatsP0DroppedSpans = new LongAdder();
- private final LongAdder clientStatsP0DroppedTraces = new LongAdder();
- private final LongAdder clientStatsRequests = new LongAdder();
- private final LongAdder clientStatsErrors = new LongAdder();
- private final LongAdder clientStatsDowngrades = new LongAdder();
-
- private final LongAdder statsAggregateDropped = new LongAdder();
- private final LongAdder statsInboxFull = new LongAdder();
+ private final Accumulator metricAccumulator = Accumulator.of(Metric.class);
+ private volatile Accumulator.Counts storedTotal = Accumulator.Counts.zero(Metric.class);
private final StatsDClient statsd;
private final long interval;
@@ -140,21 +74,21 @@ public void onShutdown(final boolean flushSuccess) {}
public void onPublish(final List trace, final int samplingPriority) {
switch (samplingPriority) {
case USER_DROP:
- userDropEnqueuedTraces.increment();
+ metricAccumulator.inc(Metric.USER_DROP_ENQUEUED_TRACES);
break;
case USER_KEEP:
- userKeepEnqueuedTraces.increment();
+ metricAccumulator.inc(Metric.USER_KEEP_ENQUEUED_TRACES);
break;
case SAMPLER_DROP:
- samplerDropEnqueuedTraces.increment();
+ metricAccumulator.inc(Metric.SAMPLER_DROP_ENQUEUED_TRACES);
break;
case SAMPLER_KEEP:
- samplerKeepEnqueuedTraces.increment();
+ metricAccumulator.inc(Metric.SAMPLER_KEEP_ENQUEUED_TRACES);
break;
default:
- unsetPriorityEnqueuedTraces.increment();
+ metricAccumulator.inc(Metric.UNSET_PRIORITY_ENQUEUED_TRACES);
}
- enqueuedSpans.add(trace.size());
+ metricAccumulator.add(Metric.ENQUEUED_SPANS, trace.size());
checkForClientSpansWithoutContext(trace);
}
@@ -163,7 +97,7 @@ private void checkForClientSpansWithoutContext(final List trace) {
if (span != null && span.getParentId() == ZERO) {
String spanKind = span.getTag(SPAN_KIND, "undefined");
if (SPAN_KIND_CLIENT.equals(spanKind)) {
- this.clientSpansWithoutContext.increment();
+ metricAccumulator.inc(Metric.CLIENT_SPANS_WITHOUT_CONTEXT);
}
}
}
@@ -173,31 +107,31 @@ private void checkForClientSpansWithoutContext(final List trace) {
public void onFailedPublish(final int samplingPriority, final int spanCount) {
switch (samplingPriority) {
case USER_DROP:
- userDropDroppedSpans.add(spanCount);
- userDropDroppedTraces.increment();
+ metricAccumulator.add(Metric.USER_DROP_DROPPED_SPANS, spanCount);
+ metricAccumulator.inc(Metric.USER_DROP_DROPPED_TRACES);
break;
case USER_KEEP:
- userKeepDroppedSpans.add(spanCount);
- userKeepDroppedTraces.increment();
+ metricAccumulator.add(Metric.USER_KEEP_DROPPED_SPANS, spanCount);
+ metricAccumulator.inc(Metric.USER_KEEP_DROPPED_TRACES);
break;
case SAMPLER_DROP:
- samplerDropDroppedSpans.add(spanCount);
- samplerDropDroppedTraces.increment();
+ metricAccumulator.add(Metric.SAMPLER_DROP_DROPPED_SPANS, spanCount);
+ metricAccumulator.inc(Metric.SAMPLER_DROP_DROPPED_TRACES);
break;
case SAMPLER_KEEP:
- samplerKeepDroppedSpans.add(spanCount);
- samplerKeepDroppedTraces.increment();
+ metricAccumulator.add(Metric.SAMPLER_KEEP_DROPPED_SPANS, spanCount);
+ metricAccumulator.inc(Metric.SAMPLER_KEEP_DROPPED_TRACES);
break;
default:
- unsetPriorityDroppedSpans.add(spanCount);
- unsetPriorityDroppedTraces.increment();
+ metricAccumulator.add(Metric.UNSET_PRIORITY_DROPPED_SPANS, spanCount);
+ metricAccumulator.inc(Metric.UNSET_PRIORITY_DROPPED_TRACES);
}
}
@Override
public void onPartialPublish(final int numberOfDroppedSpans) {
- partialTraces.increment();
- samplerDropDroppedSpans.add(numberOfDroppedSpans);
+ metricAccumulator.inc(Metric.PARTIAL_TRACES);
+ metricAccumulator.add(Metric.SAMPLER_DROP_DROPPED_SPANS, numberOfDroppedSpans);
}
@Override
@@ -210,95 +144,97 @@ public void onFlush(final boolean early) {}
@Override
public void onPartialFlush(final int sizeInBytes) {
- partialBytes.add(sizeInBytes);
+ metricAccumulator.add(Metric.PARTIAL_BYTES, sizeInBytes);
}
@Override
public void onSingleSpanSample() {
- singleSpanSampled.increment();
+ metricAccumulator.inc(Metric.SINGLE_SPAN_SAMPLED);
}
@Override
public void onSingleSpanUnsampled() {
- singleSpanUnsampled.increment();
+ metricAccumulator.inc(Metric.SINGLE_SPAN_UNSAMPLED);
}
@Override
public void onSerialize(final int serializedSizeInBytes) {
// DQH - Because of Java tracer's 2 phase acceptance and serialization scheme, this doesn't
// map precisely
- enqueuedBytes.add(serializedSizeInBytes);
+ metricAccumulator.add(Metric.ENQUEUED_BYTES, serializedSizeInBytes);
}
@Override
public void onFailedSerialize(final List trace, final Throwable optionalCause) {
if (trace != null) {
- serialFailedDroppedTraces.increment();
- serialFailedDroppedSpans.add(trace.size());
+ metricAccumulator.inc(Metric.SERIAL_FAILED_DROPPED_TRACES);
+ metricAccumulator.add(Metric.SERIAL_FAILED_DROPPED_SPANS, trace.size());
}
}
@Override
public void onCreateSpan() {
- createdSpans.increment();
+ metricAccumulator.inc(Metric.CREATED_SPANS);
}
@Override
public void onFinishSpan() {
- finishedSpans.increment();
+ metricAccumulator.inc(Metric.FINISHED_SPANS);
}
@Override
public void onCreateTrace() {
- createdTraces.increment();
+ metricAccumulator.inc(Metric.CREATED_TRACES);
}
@Override
public void onScopeCloseError(boolean manual) {
- scopeCloseErrors.increment();
if (manual) {
- userScopeCloseErrors.increment();
+ metricAccumulator.inc(Metric.SCOPE_CLOSE_ERRORS);
+ metricAccumulator.inc(Metric.USER_SCOPE_CLOSE_ERRORS);
+ } else {
+ metricAccumulator.inc(Metric.SCOPE_CLOSE_ERRORS);
}
}
@Override
public void onCaptureContinuation() {
- capturedContinuations.increment();
+ metricAccumulator.inc(Metric.CAPTURED_CONTINUATIONS);
}
@Override
public void onCancelContinuation() {
- cancelledContinuations.increment();
+ metricAccumulator.inc(Metric.CANCELLED_CONTINUATIONS);
}
@Override
public void onFinishContinuation() {
- finishedContinuations.increment();
+ metricAccumulator.inc(Metric.FINISHED_CONTINUATIONS);
}
@Override
public void onActivateScope() {
- activatedScopes.increment();
+ metricAccumulator.inc(Metric.ACTIVATED_SCOPES);
}
@Override
public void onCloseScope() {
- closedScopes.increment();
+ metricAccumulator.inc(Metric.CLOSED_SCOPES);
}
@Override
public void onScopeStackOverflow() {
- scopeStackOverflow.increment();
+ metricAccumulator.inc(Metric.SCOPE_STACK_OVERFLOW);
}
@Override
public void onOrgGuardEnforce(OrgGuard.Reason reason) {
switch (reason) {
case MISMATCH:
- orgGuardEnforceMismatch.increment();
+ metricAccumulator.inc(Metric.ORG_GUARD_ENFORCE_MISMATCH);
break;
case STRICT_MISSING:
- orgGuardEnforceStrictMissing.increment();
+ metricAccumulator.inc(Metric.ORG_GUARD_ENFORCE_STRICT_MISSING);
break;
}
}
@@ -317,68 +253,68 @@ public void onFailedSend(
@Override
public void onLongRunningUpdate(final int dropped, final int write, final int expired) {
- longRunningTracesWrite.add(write);
- longRunningTracesDropped.add(dropped);
- longRunningTracesExpired.add(expired);
+ metricAccumulator.add(Metric.LONG_RUNNING_TRACES_WRITE, write);
+ metricAccumulator.add(Metric.LONG_RUNNING_TRACES_DROPPED, dropped);
+ metricAccumulator.add(Metric.LONG_RUNNING_TRACES_EXPIRED, expired);
}
private void onSendAttempt(
final int traceCount, final int sizeInBytes, final RemoteApi.Response response) {
- apiRequests.increment();
- flushedTraces.add(traceCount);
+ metricAccumulator.inc(Metric.API_REQUESTS);
+ metricAccumulator.add(Metric.FLUSHED_TRACES, traceCount);
// TODO: missing queue.spans (# of spans being sent)
- flushedBytes.add(sizeInBytes);
+ metricAccumulator.add(Metric.FLUSHED_BYTES, sizeInBytes);
if (response.exception().isPresent()) {
// covers communication errors -- both not receiving a response or
// receiving malformed response (even when otherwise successful)
- apiErrors.increment();
+ metricAccumulator.inc(Metric.API_ERRORS);
}
- int status = response.status().orElse(0);
- if (status != 0) {
- if (200 == status) {
- apiResponsesOK.increment();
- } else {
- statsd.incrementCounter("api.responses.total", statusTagsCache.get(status));
- }
+ if (200 == response.status().orElse(0)) {
+ metricAccumulator.inc(Metric.API_RESPONSES_OK);
+ }
+
+ final int status = response.status().orElse(0);
+ if (status != 0 && 200 != status) {
+ statsd.incrementCounter("api.responses.total", statusTagsCache.get(status));
}
}
@Override
public void onClientStatTraceComputed(int countedSpans, int totalSpans, boolean dropped) {
- clientStatsProcessedTraces.increment();
- clientStatsProcessedSpans.add(countedSpans);
+ metricAccumulator.inc(Metric.CLIENT_STATS_PROCESSED_TRACES);
+ metricAccumulator.add(Metric.CLIENT_STATS_PROCESSED_SPANS, countedSpans);
if (dropped) {
- clientStatsP0DroppedTraces.increment();
- clientStatsP0DroppedSpans.add(totalSpans);
+ metricAccumulator.inc(Metric.CLIENT_STATS_P0_DROPPED_TRACES);
+ metricAccumulator.add(Metric.CLIENT_STATS_P0_DROPPED_SPANS, totalSpans);
}
}
@Override
public void onClientStatPayloadSent() {
- clientStatsRequests.increment();
+ metricAccumulator.inc(Metric.CLIENT_STATS_REQUESTS);
}
@Override
public void onClientStatDowngraded() {
- clientStatsDowngrades.increment();
+ metricAccumulator.inc(Metric.CLIENT_STATS_DOWNGRADES);
}
@Override
public void onClientStatErrorReceived() {
- clientStatsErrors.increment();
+ metricAccumulator.inc(Metric.CLIENT_STATS_ERRORS);
}
@Override
public void onStatsAggregateDropped() {
- statsAggregateDropped.increment();
+ metricAccumulator.inc(Metric.STATS_AGGREGATE_DROPPED);
statsd.count("datadog.tracer.stats.collapsed_spans", 1, COLLAPSED_WHOLE_KEY_TAGS);
}
@Override
public void onStatsInboxFull() {
- statsInboxFull.increment();
+ metricAccumulator.inc(Metric.STATS_INBOX_FULL);
}
@Override
@@ -395,299 +331,139 @@ public void close() {
private static class Flush implements AgentTaskScheduler.Task {
- private static final String[] USER_DROP_TAG = new String[] {"priority:user_drop"};
- private static final String[] USER_KEEP_TAG = new String[] {"priority:user_keep"};
- private static final String[] SAMPLER_DROP_TAG = new String[] {"priority:sampler_drop"};
- private static final String[] SAMPLER_KEEP_TAG = new String[] {"priority:sampler_keep"};
- private static final String[] SERIAL_FAILED_TAG = new String[] {"failure:serial"};
- private static final String[] UNSET_TAG = new String[] {"priority:unset"};
- private static final String[] SINGLE_SPAN_SAMPLER = new String[] {"sampler:single-span"};
- private static final String[] REASON_LRU_EVICTION_TAG = new String[] {"reason:lru_eviction"};
- private static final String[] REASON_INBOX_FULL_TAG = new String[] {"reason:inbox_full"};
- private static final String[] ORG_GUARD_MISMATCH_TAGS = new String[] {"reason:mismatch"};
- private static final String[] ORG_GUARD_STRICT_MISSING_TAGS =
- new String[] {"reason:strict_missing"};
-
- private final long[] previousCounts = new long[54];
-
- @SuppressFBWarnings("AT_STALE_THREAD_WRITE_OF_PRIMITIVE")
- private int countIndex;
-
@Override
public void run(TracerHealthMetrics target) {
- countIndex = -1; // reposition so _next_ value is 0
- try {
-
- reportIfChanged(target.statsd, "api.requests.total", target.apiRequests, NO_TAGS);
- reportIfChanged(target.statsd, "api.errors.total", target.apiErrors, NO_TAGS);
- // non-OK responses are reported immediately in onSendAttempt with different status tags
- reportIfChanged(
- target.statsd, "api.responses.total", target.apiResponsesOK, STATUS_OK_TAGS);
-
- reportIfChanged(
- target.statsd, "queue.enqueued.traces", target.userDropEnqueuedTraces, USER_DROP_TAG);
- reportIfChanged(
- target.statsd, "queue.enqueued.traces", target.userKeepEnqueuedTraces, USER_KEEP_TAG);
- reportIfChanged(
- target.statsd,
- "queue.enqueued.traces",
- target.samplerDropEnqueuedTraces,
- SAMPLER_DROP_TAG);
- reportIfChanged(
- target.statsd,
- "queue.enqueued.traces",
- target.samplerKeepEnqueuedTraces,
- SAMPLER_KEEP_TAG);
- reportIfChanged(
- target.statsd, "queue.enqueued.traces", target.unsetPriorityEnqueuedTraces, UNSET_TAG);
-
- reportIfChanged(
- target.statsd, "queue.dropped.traces", target.userDropDroppedTraces, USER_DROP_TAG);
- reportIfChanged(
- target.statsd, "queue.dropped.traces", target.userKeepDroppedTraces, USER_KEEP_TAG);
- reportIfChanged(
- target.statsd,
- "queue.dropped.traces",
- target.samplerDropDroppedTraces,
- SAMPLER_DROP_TAG);
- reportIfChanged(
- target.statsd,
- "queue.dropped.traces",
- target.samplerKeepDroppedTraces,
- SAMPLER_KEEP_TAG);
- reportIfChanged(
- target.statsd,
- "queue.dropped.traces",
- target.serialFailedDroppedTraces,
- SERIAL_FAILED_TAG);
- reportIfChanged(
- target.statsd, "queue.dropped.traces", target.unsetPriorityDroppedTraces, UNSET_TAG);
-
- reportIfChanged(
- target.statsd, "queue.dropped.spans", target.userDropDroppedSpans, USER_DROP_TAG);
- reportIfChanged(
- target.statsd, "queue.dropped.spans", target.userKeepDroppedSpans, USER_KEEP_TAG);
- reportIfChanged(
- target.statsd, "queue.dropped.spans", target.samplerDropDroppedSpans, SAMPLER_DROP_TAG);
- reportIfChanged(
- target.statsd, "queue.dropped.spans", target.samplerKeepDroppedSpans, SAMPLER_KEEP_TAG);
- reportIfChanged(
- target.statsd,
- "queue.dropped.spans",
- target.serialFailedDroppedSpans,
- SERIAL_FAILED_TAG);
- reportIfChanged(
- target.statsd, "queue.dropped.spans", target.unsetPriorityDroppedSpans, UNSET_TAG);
-
- reportIfChanged(target.statsd, "queue.enqueued.spans", target.enqueuedSpans, NO_TAGS);
- reportIfChanged(target.statsd, "queue.enqueued.bytes", target.enqueuedBytes, NO_TAGS);
- reportIfChanged(target.statsd, "trace.pending.created", target.createdTraces, NO_TAGS);
- reportIfChanged(target.statsd, "span.pending.created", target.createdSpans, NO_TAGS);
- reportIfChanged(target.statsd, "span.pending.finished", target.finishedSpans, NO_TAGS);
- reportIfChanged(target.statsd, "flush.traces.total", target.flushedTraces, NO_TAGS);
- reportIfChanged(target.statsd, "flush.bytes.total", target.flushedBytes, NO_TAGS);
- reportIfChanged(target.statsd, "queue.partial.traces", target.partialTraces, NO_TAGS);
- reportIfChanged(target.statsd, "span.flushed.partial", target.partialBytes, NO_TAGS);
- reportIfChanged(
- target.statsd, "span.client.no-context", target.clientSpansWithoutContext, NO_TAGS);
-
- reportIfChanged(
- target.statsd, "span.sampling.sampled", target.singleSpanSampled, SINGLE_SPAN_SAMPLER);
- reportIfChanged(
- target.statsd,
- "span.sampling.unsampled",
- target.singleSpanUnsampled,
- SINGLE_SPAN_SAMPLER);
-
- reportIfChanged(
- target.statsd, "span.continuations.captured", target.capturedContinuations, NO_TAGS);
- reportIfChanged(
- target.statsd, "span.continuations.canceled", target.cancelledContinuations, NO_TAGS);
- reportIfChanged(
- target.statsd, "span.continuations.finished", target.finishedContinuations, NO_TAGS);
-
- reportIfChanged(target.statsd, "scope.activate.count", target.activatedScopes, NO_TAGS);
- reportIfChanged(target.statsd, "scope.close.count", target.closedScopes, NO_TAGS);
- reportIfChanged(
- target.statsd, "scope.error.stack-overflow", target.scopeStackOverflow, NO_TAGS);
- reportIfChanged(target.statsd, "scope.close.error", target.scopeCloseErrors, NO_TAGS);
- reportIfChanged(
- target.statsd, "scope.user.close.error", target.userScopeCloseErrors, NO_TAGS);
-
- reportIfChanged(
- target.statsd, "long-running.write", target.longRunningTracesWrite, NO_TAGS);
- reportIfChanged(
- target.statsd, "long-running.dropped", target.longRunningTracesDropped, NO_TAGS);
- reportIfChanged(
- target.statsd, "long-running.expired", target.longRunningTracesExpired, NO_TAGS);
-
- reportIfChanged(
- target.statsd,
- "org_guard.enforce",
- target.orgGuardEnforceMismatch,
- ORG_GUARD_MISMATCH_TAGS);
- reportIfChanged(
- target.statsd,
- "org_guard.enforce",
- target.orgGuardEnforceStrictMissing,
- ORG_GUARD_STRICT_MISSING_TAGS);
-
- reportIfChanged(
- target.statsd, "stats.traces_in", target.clientStatsProcessedTraces, NO_TAGS);
- reportIfChanged(target.statsd, "stats.spans_in", target.clientStatsProcessedSpans, NO_TAGS);
- reportIfChanged(
- target.statsd, "stats.dropped_p0_traces", target.clientStatsP0DroppedTraces, NO_TAGS);
- reportIfChanged(
- target.statsd, "stats.dropped_p0_spans", target.clientStatsP0DroppedSpans, NO_TAGS);
- reportIfChanged(target.statsd, "stats.flush_payloads", target.clientStatsRequests, NO_TAGS);
- reportIfChanged(target.statsd, "stats.flush_errors", target.clientStatsErrors, NO_TAGS);
- reportIfChanged(
- target.statsd, "stats.agent_downgrades", target.clientStatsDowngrades, NO_TAGS);
- reportIfChanged(
- target.statsd,
- "stats.dropped_aggregates",
- target.statsAggregateDropped,
- REASON_LRU_EVICTION_TAG);
- reportIfChanged(
- target.statsd,
- "stats.dropped_aggregates",
- target.statsInboxFull,
- REASON_INBOX_FULL_TAG);
-
- } catch (ArrayIndexOutOfBoundsException e) {
- log.warn(
- "previousCounts array needs resizing to at least {}, was {}",
- countIndex + 1,
- previousCounts.length);
- }
- }
-
- private void reportIfChanged(
- StatsDClient statsDClient, String aspect, LongAdder counter, String[] tags) {
- long count = counter.sum();
- long delta = count - previousCounts[++countIndex];
- if (delta > 0) {
- statsDClient.count(aspect, delta, tags);
- previousCounts[countIndex] = count;
- }
+ Accumulator.Counts delta = target.metricAccumulator.accumulateAndReset();
+ StatsDCountReporter.report(target.statsd, delta);
+ target.storedTotal = target.storedTotal.plus(delta);
}
}
@Override
public String summary() {
- return "apiRequests="
- + apiRequests.sum()
- + "\napiErrors="
- + apiErrors.sum()
- + "\napiResponsesOK="
- + apiResponsesOK.sum()
- + "\n"
- + "\nuserDropEnqueuedTraces="
- + userDropEnqueuedTraces.sum()
- + "\nuserKeepEnqueuedTraces="
- + userKeepEnqueuedTraces.sum()
- + "\nsamplerDropEnqueuedTraces="
- + samplerDropEnqueuedTraces.sum()
- + "\nsamplerKeepEnqueuedTraces="
- + samplerKeepEnqueuedTraces.sum()
- + "\nunsetPriorityEnqueuedTraces="
- + unsetPriorityEnqueuedTraces.sum()
- + "\n"
- + "\nuserDropDroppedTraces="
- + userDropDroppedTraces.sum()
- + "\nuserKeepDroppedTraces="
- + userKeepDroppedTraces.sum()
- + "\nsamplerDropDroppedTraces="
- + samplerDropDroppedTraces.sum()
- + "\nsamplerKeepDroppedTraces="
- + samplerKeepDroppedTraces.sum()
- + "\nserialFailedDroppedTraces="
- + serialFailedDroppedTraces.sum()
- + "\nunsetPriorityDroppedTraces="
- + unsetPriorityDroppedTraces.sum()
- + "\n"
- + "\nuserDropDroppedSpans="
- + userDropDroppedSpans.sum()
- + "\nuserKeepDroppedSpans="
- + userKeepDroppedSpans.sum()
- + "\nsamplerDropDroppedSpans="
- + samplerDropDroppedSpans.sum()
- + "\nsamplerKeepDroppedSpans="
- + samplerKeepDroppedSpans.sum()
- + "\nserialFailedDroppedSpans="
- + serialFailedDroppedSpans.sum()
- + "\nunsetPriorityDroppedSpans="
- + unsetPriorityDroppedSpans.sum()
- + "\n"
- + "\nenqueuedSpans="
- + enqueuedSpans.sum()
- + "\nenqueuedBytes="
- + enqueuedBytes.sum()
- + "\ncreatedTraces="
- + createdTraces.sum()
- + "\ncreatedSpans="
- + createdSpans.sum()
- + "\nfinishedSpans="
- + finishedSpans.sum()
- + "\nflushedTraces="
- + flushedTraces.sum()
- + "\nflushedBytes="
- + flushedBytes.sum()
- + "\npartialTraces="
- + partialTraces.sum()
- + "\npartialBytes="
- + partialBytes.sum()
- + "\n"
- + "\nclientSpansWithoutContext="
- + clientSpansWithoutContext.sum()
- + "\n"
- + "\nsingleSpanSampled="
- + singleSpanSampled.sum()
- + "\nsingleSpanUnsampled="
- + singleSpanUnsampled.sum()
- + "\n"
- + "\ncapturedContinuations="
- + capturedContinuations.sum()
- + "\ncancelledContinuations="
- + cancelledContinuations.sum()
- + "\nfinishedContinuations="
- + finishedContinuations.sum()
- + "\n"
- + "\nactivatedScopes="
- + activatedScopes.sum()
- + "\nclosedScopes="
- + closedScopes.sum()
- + "\nscopeStackOverflow="
- + scopeStackOverflow.sum()
- + "\nscopeCloseErrors="
- + scopeCloseErrors.sum()
- + "\nuserScopeCloseErrors="
- + userScopeCloseErrors.sum()
- + "\n"
- + "\nlongRunningTracesWrite="
- + longRunningTracesWrite.sum()
- + "\nlongRunningTracesDropped="
- + longRunningTracesDropped.sum()
- + "\nlongRunningTracesExpired="
- + longRunningTracesExpired.sum()
- + "\n"
- + "\nclientStatsRequests="
- + clientStatsRequests.sum()
- + "\nclientStatsErrors="
- + clientStatsErrors.sum()
- + "\nclientStatsDowngrades="
- + clientStatsDowngrades.sum()
- + "\nclientStatsP0DroppedSpans="
- + clientStatsP0DroppedSpans.sum()
- + "\nclientStatsP0DroppedTraces="
- + clientStatsP0DroppedTraces.sum()
- + "\nclientStatsProcessedSpans="
- + clientStatsProcessedSpans.sum()
- + "\nclientStatsProcessedTraces="
- + clientStatsProcessedTraces.sum()
- + "\nstatsAggregateDropped="
- + statsAggregateDropped.sum()
- + "\nstatsInboxFull="
- + statsInboxFull.sum();
+ Accumulator.Counts live = storedTotal.plus(metricAccumulator.sum());
+ StringBuilder summary = new StringBuilder();
+ for (Metric metric : live.keys()) {
+ if (summary.length() > 0) {
+ summary.append('\n');
+ }
+ summary.append(metric.getSummaryLabel()).append('=').append(live.get(metric));
+ }
+ return summary.toString();
+ }
+
+ /**
+ * One counter tracked by {@link TracerHealthMetrics}: a dogstatsd metric name + tags, plus the
+ * label {@link TracerHealthMetrics#summary()} renders it under. One constant per (counter, tag)
+ * combination -- several constants can share a metric name but differ by tag, mirroring the
+ * distinct {@code LongAdder} fields this enum replaces.
+ */
+ enum Metric implements StatsDCounterKey {
+ API_REQUESTS("apiRequests", "api.requests.total"),
+ API_ERRORS("apiErrors", "api.errors.total"),
+ // non-OK responses are reported immediately in onSendAttempt with different status tags
+ API_RESPONSES_OK("apiResponsesOK", "api.responses.total", "status:200"),
+
+ USER_DROP_ENQUEUED_TRACES(
+ "userDropEnqueuedTraces", "queue.enqueued.traces", "priority:user_drop"),
+ USER_KEEP_ENQUEUED_TRACES(
+ "userKeepEnqueuedTraces", "queue.enqueued.traces", "priority:user_keep"),
+ SAMPLER_DROP_ENQUEUED_TRACES(
+ "samplerDropEnqueuedTraces", "queue.enqueued.traces", "priority:sampler_drop"),
+ SAMPLER_KEEP_ENQUEUED_TRACES(
+ "samplerKeepEnqueuedTraces", "queue.enqueued.traces", "priority:sampler_keep"),
+ UNSET_PRIORITY_ENQUEUED_TRACES(
+ "unsetPriorityEnqueuedTraces", "queue.enqueued.traces", "priority:unset"),
+
+ USER_DROP_DROPPED_TRACES("userDropDroppedTraces", "queue.dropped.traces", "priority:user_drop"),
+ USER_KEEP_DROPPED_TRACES("userKeepDroppedTraces", "queue.dropped.traces", "priority:user_keep"),
+ SAMPLER_DROP_DROPPED_TRACES(
+ "samplerDropDroppedTraces", "queue.dropped.traces", "priority:sampler_drop"),
+ SAMPLER_KEEP_DROPPED_TRACES(
+ "samplerKeepDroppedTraces", "queue.dropped.traces", "priority:sampler_keep"),
+ SERIAL_FAILED_DROPPED_TRACES(
+ "serialFailedDroppedTraces", "queue.dropped.traces", "failure:serial"),
+ UNSET_PRIORITY_DROPPED_TRACES(
+ "unsetPriorityDroppedTraces", "queue.dropped.traces", "priority:unset"),
+
+ USER_DROP_DROPPED_SPANS("userDropDroppedSpans", "queue.dropped.spans", "priority:user_drop"),
+ USER_KEEP_DROPPED_SPANS("userKeepDroppedSpans", "queue.dropped.spans", "priority:user_keep"),
+ SAMPLER_DROP_DROPPED_SPANS(
+ "samplerDropDroppedSpans", "queue.dropped.spans", "priority:sampler_drop"),
+ SAMPLER_KEEP_DROPPED_SPANS(
+ "samplerKeepDroppedSpans", "queue.dropped.spans", "priority:sampler_keep"),
+ SERIAL_FAILED_DROPPED_SPANS(
+ "serialFailedDroppedSpans", "queue.dropped.spans", "failure:serial"),
+ UNSET_PRIORITY_DROPPED_SPANS(
+ "unsetPriorityDroppedSpans", "queue.dropped.spans", "priority:unset"),
+
+ ENQUEUED_SPANS("enqueuedSpans", "queue.enqueued.spans"),
+ ENQUEUED_BYTES("enqueuedBytes", "queue.enqueued.bytes"),
+ CREATED_TRACES("createdTraces", "trace.pending.created"),
+ CREATED_SPANS("createdSpans", "span.pending.created"),
+ FINISHED_SPANS("finishedSpans", "span.pending.finished"),
+ FLUSHED_TRACES("flushedTraces", "flush.traces.total"),
+ FLUSHED_BYTES("flushedBytes", "flush.bytes.total"),
+ PARTIAL_TRACES("partialTraces", "queue.partial.traces"),
+ PARTIAL_BYTES("partialBytes", "span.flushed.partial"),
+ CLIENT_SPANS_WITHOUT_CONTEXT("clientSpansWithoutContext", "span.client.no-context"),
+
+ SINGLE_SPAN_SAMPLED("singleSpanSampled", "span.sampling.sampled", "sampler:single-span"),
+ SINGLE_SPAN_UNSAMPLED("singleSpanUnsampled", "span.sampling.unsampled", "sampler:single-span"),
+
+ CAPTURED_CONTINUATIONS("capturedContinuations", "span.continuations.captured"),
+ CANCELLED_CONTINUATIONS("cancelledContinuations", "span.continuations.canceled"),
+ FINISHED_CONTINUATIONS("finishedContinuations", "span.continuations.finished"),
+
+ ACTIVATED_SCOPES("activatedScopes", "scope.activate.count"),
+ CLOSED_SCOPES("closedScopes", "scope.close.count"),
+ SCOPE_STACK_OVERFLOW("scopeStackOverflow", "scope.error.stack-overflow"),
+ SCOPE_CLOSE_ERRORS("scopeCloseErrors", "scope.close.error"),
+ USER_SCOPE_CLOSE_ERRORS("userScopeCloseErrors", "scope.user.close.error"),
+
+ LONG_RUNNING_TRACES_WRITE("longRunningTracesWrite", "long-running.write"),
+ LONG_RUNNING_TRACES_DROPPED("longRunningTracesDropped", "long-running.dropped"),
+ LONG_RUNNING_TRACES_EXPIRED("longRunningTracesExpired", "long-running.expired"),
+
+ ORG_GUARD_ENFORCE_MISMATCH("orgGuardEnforceMismatch", "org_guard.enforce", "reason:mismatch"),
+ ORG_GUARD_ENFORCE_STRICT_MISSING(
+ "orgGuardEnforceStrictMissing", "org_guard.enforce", "reason:strict_missing"),
+
+ CLIENT_STATS_PROCESSED_TRACES("clientStatsProcessedTraces", "stats.traces_in"),
+ CLIENT_STATS_PROCESSED_SPANS("clientStatsProcessedSpans", "stats.spans_in"),
+ CLIENT_STATS_P0_DROPPED_TRACES("clientStatsP0DroppedTraces", "stats.dropped_p0_traces"),
+ CLIENT_STATS_P0_DROPPED_SPANS("clientStatsP0DroppedSpans", "stats.dropped_p0_spans"),
+ CLIENT_STATS_REQUESTS("clientStatsRequests", "stats.flush_payloads"),
+ CLIENT_STATS_ERRORS("clientStatsErrors", "stats.flush_errors"),
+ CLIENT_STATS_DOWNGRADES("clientStatsDowngrades", "stats.agent_downgrades"),
+
+ STATS_AGGREGATE_DROPPED(
+ "statsAggregateDropped", "stats.dropped_aggregates", "reason:lru_eviction"),
+ STATS_INBOX_FULL("statsInboxFull", "stats.dropped_aggregates", "reason:inbox_full"),
+ ;
+
+ private final String summaryLabel;
+ private final String metricName;
+ private final String[] tags;
+
+ Metric(String summaryLabel, String metricName, String... tags) {
+ this.summaryLabel = summaryLabel;
+ this.metricName = metricName;
+ this.tags = tags;
+ }
+
+ @Override
+ public String getMetricName() {
+ return metricName;
+ }
+
+ @Override
+ public String[] getTags() {
+ return tags;
+ }
+
+ String getSummaryLabel() {
+ return summaryLabel;
+ }
}
}
diff --git a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java
index 3b7c7eeb05f..5345f490096 100644
--- a/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java
+++ b/internal-api/src/jmh/java/datadog/trace/util/AccumulatorBenchmark.java
@@ -3,6 +3,7 @@
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import org.openjdk.jmh.annotations.Benchmark;
@@ -20,128 +21,135 @@
import org.openjdk.jmh.infra.Blackhole;
/**
- * {@link Accumulator} vs {@link LongAdder} vs the {@code ConcurrentHashMap.computeIfAbsent(key, k
- * -> new AtomicLong())} anti-pattern, at one thread (no contention) and at {@link Threads#MAX}
- * (heavy contention). The CHM variant allocates its counter under the bucket's bin lock the first
- * time its one constant key is seen -- exactly the pathology {@link Accumulator} exists to avoid --
- * but since the map is a {@code @State(Scope.Benchmark)} field shared across the whole run, that
- * allocation happens exactly once; every sampled op after it hits the warmed, already-present fast
- * path. So this measures steady-state {@code computeIfAbsent} lookup overhead on an
- * already-populated map, not the one-time allocation-under-lock cost -- still a useful number (a
- * fixed, small key set that's allocated once and hit for the life of the process, as {@code
+ * {@link Accumulator} vs the alternatives it actually displaces: a single {@code LongAdder} (the
+ * collision-free baseline it can never beat, only approach), an independent {@code LongAdder} per
+ * counter guarded by a per-counter lock (the "just fix it with LongAdder" natural migration target
+ * -- {@code longAdderGroup*}), and the {@code ConcurrentHashMap.computeIfAbsent(key, k -> new
+ * AtomicLong())} anti-pattern ({@code chmAtomicLongIncrement*}) that {@link Accumulator} exists to
+ * avoid. The CHM variant allocates its counter under the bucket's bin lock the first time its one
+ * constant key is seen, but since the map is a {@code @State(Scope.Benchmark)} field shared across
+ * the whole run, that allocation happens exactly once; every sampled op after it hits the warmed,
+ * already-present fast path. So this measures steady-state {@code computeIfAbsent} lookup overhead
+ * on an already-populated map, not the one-time allocation-under-lock cost -- still a useful number
+ * (a fixed, small key set that's allocated once and hit for the life of the process, as {@code
* WafMetricCollector}-style CHM counters are, spends nearly all its time in this same warmed path),
* just not the pathology the name of this benchmark might suggest.
*
- * Contention result to note: at low contention, {@code accumulatorIncrement} is
- * essentially free and on par with {@code longAdderIncrement}. At {@code Threads.MAX} (10 threads
- * on the measurement machine), oversizing {@link Accumulator}'s stripe count from 8 (one per core)
- * to 16 (roughly 2x cores, see {@code stripeCount()}) cut {@code
- * accumulatorIncrement_highContention} from ~0.097 us/op to ~0.040 us/op -- fewer threads collide
- * on a stripe, so fewer of them pay {@code synchronized}'s blocking wait instead of a cheap
- * fast-path lock. It is still roughly 4-5x slower than {@code longAdderIncrement} (a collision-free
- * CAS retry beats even an uncontended monitor enter/exit), and {@code accumulateAndReset} under
- * concurrent writers got correspondingly more expensive (~7.5us to ~15.5us) since draining now
- * walks twice as many stripes while writers are actively landing on them. Read {@code
- * accumulatorIncrement_highContention} not as "Accumulator beats LongAdder under contention" (it
- * doesn't, on this shape) but as the honest cost of the drain-under-lock design that buys atomic
- * combine+reset; a caller trading that safety for raw increment throughput should measure their own
- * contention level before choosing between them.
- * Apple M1 Max, 10 CPUs - JDK 1.8.0_382 (Zulu) - macOS/arm64 - stripeCount() = 16
- * Benchmark Mode Cnt Score Error Units
- * AccumulatorBenchmark.longAdderIncrement_lowContention avgt 6 0.007 ± 0.001 us/op
- * AccumulatorBenchmark.longAdderIncrement_highContention avgt 6 0.009 ± 0.001 us/op
- * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 6 0.010 ± 0.001 us/op
- * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.040 ± 0.002 us/op
- * AccumulatorBenchmark.chmAtomicLongIncrement_lowContention avgt 6 0.010 ± 0.001 us/op
- * AccumulatorBenchmark.chmAtomicLongIncrement_highContention avgt 6 0.417 ± 0.543 us/op
- * AccumulatorBenchmark.longAdderSumThenReset_lowContention avgt 6 0.012 ± 0.001 us/op
- * AccumulatorBenchmark.longAdderSumThenReset_highContention avgt 6 2.433 ± 0.203 us/op
- * AccumulatorBenchmark.accumulatorAccumulateAndReset_lowContention avgt 6 0.162 ± 0.009 us/op
- * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 15.515 ± 4.094 us/op
- *
- *
- *
(This run had some background noise from another session on the measurement machine; the
- * {@code lowContention} rows and the {@code highContention} directional deltas are reliable, but
- * treat the exact {@code highContention} magnitudes as approximate.)
- *
*
{@code longAdderGroup*}: is a "just fix it with LongAdder" helper actually cheaper?
* {@code groupInc}/{@code groupAccumulateAnd} are the natural correct fix using {@code LongAdder}
* as the payload: one {@code LongAdder} per counter, with a per-counter lock guarding both
* the increment and the drain (locking only the drain does nothing -- {@code sumThenReset()}'s
* internal race is against the {@code LongAdder}'s own CAS-based {@code add()}, not against any
- * lock a caller takes). This closes the same reset hazard as {@link Accumulator}, but stripes by
- * counter instead of by thread.
- * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.029 ± 0.051 us/op
- * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 13.431 ± 5.876 us/op
- * AccumulatorBenchmark.longAdderGroupIncrement_highContention avgt 6 0.294 ± 0.088 us/op
- * AccumulatorBenchmark.longAdderGroupAccumulateAnd_highContention avgt 6 0.549 ± 0.291 us/op
- * Not "similar cost" -- a clean trade-off inversion. With this benchmark's single counter,
- * {@code longAdderGroup}'s per-counter lock collapses to one lock for every thread (no thread-based
- * distribution at all), so it loses badly on the write path: ~10x worse than {@code Accumulator}'s
- * thread-sharded stripes. But its drain only has that one lock to acquire, so it wins big there:
- * ~24x better than {@code Accumulator}, which always walks all 16 stripes on every drain regardless
- * of counter count. That asymmetry is the whole story: {@code longAdderGroup}'s drain cost scales
- * with number of counters (more counters -> more locks to drain), while {@code
- * Accumulator}'s drain cost is fixed at stripe count, independent of counter count. Which design
- * actually wins for a given caller depends on that caller's counter cardinality and whether its
- * write traffic concentrates on a few hot counters (favors thread-sharding) or spreads across many
- * (favors counter-sharding) -- not measured here, and worth checking against the real migration
- * targets before treating either number as the general answer.
- *
- *
{@code typed*}: what does the {@link Accumulator}/{@link Accumulator.Stripe}/{@link
- * Accumulator.Counts} wrapping actually cost over calling {@link Accumulator.EmbeddingSupport}
- * directly? {@code typedIncrement}/{@code typedUpdate} pair against {@code
- * accumulatorIncrement} (the same underlying call), and {@code typedAccumulateAndReset} pairs
- * against {@code accumulatorAccumulateAndReset}.
- * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 6 0.010 ± 0.001 us/op
- * AccumulatorBenchmark.typedIncrement_lowContention avgt 6 0.010 ± 0.001 us/op
- * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 6 0.033 ± 0.008 us/op
- * AccumulatorBenchmark.typedIncrement_highContention avgt 6 0.025 ± 0.015 us/op
- * AccumulatorBenchmark.typedUpdate_lowContention avgt 6 0.010 ± 0.001 us/op
- * AccumulatorBenchmark.typedUpdate_highContention avgt 6 0.037 ± 0.017 us/op
- * AccumulatorBenchmark.accumulatorAccumulateAndReset_lowContention avgt 6 0.161 ± 0.003 us/op
- * AccumulatorBenchmark.typedAccumulateAndReset_lowContention avgt 6 0.164 ± 0.005 us/op
- * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 6 17.399 ± 3.091 us/op
- * AccumulatorBenchmark.typedAccumulateAndReset_highContention avgt 6 12.666 ± 1.272 us/op
- * {@code typedIncrement}/{@code typedUpdate} track the raw calls within noise at both
- * contention levels -- the field-load indirection through the {@link Accumulator} instance and the
- * fresh {@link Accumulator.Stripe} constructed under {@code update}'s held lock both disappear,
- * consistent with a small, non-capturing mutator letting escape analysis scalar-replace the {@code
- * Stripe}. {@code typedAccumulateAndReset} also tracks the raw drain within noise at low contention
- * (0.164 vs 0.161 us/op) -- the one-{@link Accumulator.Counts}-object-per-drain allocation it's
- * documented to pay doesn't show up at this granularity. The high-contention gap in the other
- * direction (12.666 vs 17.399) is not a real typed-vs-raw effect -- wrapping an already-drained
- * array can only add cost, never remove it -- it's the same run-to-run lock-contention noise this
- * exact measurement already shows above (13.431, 15.515, 17.399 us/op across three otherwise
- * identical runs). Net: the wrapper's cost was not measurable in this run.
+ * lock a caller takes). The single-counter {@code *_*} benchmarks below collapse that per-counter
+ * lock to one lock shared by every thread -- the degenerate worst case for {@code longAdderGroup},
+ * with no thread-based distribution at all. The {@code *8_*} benchmarks fix that: each JMH worker
+ * thread is pinned to one of 8 counters for its lifetime (see {@link #threadCounterIndex}), so
+ * {@code longAdderGroup8}'s threads split into up to 8 groups each contending their own lock --
+ * the topology where distributed locking should actually pay off, forcing {@link Accumulator}'s
+ * thread-striped design to earn its write-side win rather than facing a single-counter worst case.
+ * Fork(5), 15 samples per benchmark:
+ * AccumulatorBenchmark.accumulatorAccumulateAndReset_highContention avgt 15 2.746 ± 0.050 us/op
+ * AccumulatorBenchmark.accumulatorAccumulateAndReset_lowContention avgt 15 0.056 ± 0.001 us/op
+ * AccumulatorBenchmark.accumulatorAccumulateAndReset8_highContention avgt 15 6.875 ± 0.422 us/op
+ * AccumulatorBenchmark.accumulatorAccumulateAndReset8_lowContention avgt 15 0.363 ± 0.003 us/op
+ * AccumulatorBenchmark.accumulatorIncrement_highContention avgt 15 0.009 ± 0.001 us/op
+ * AccumulatorBenchmark.accumulatorIncrement_lowContention avgt 15 0.007 ± 0.001 us/op
+ * AccumulatorBenchmark.accumulatorIncrement8_highContention avgt 15 0.017 ± 0.009 us/op
+ * AccumulatorBenchmark.accumulatorIncrement8_lowContention avgt 15 0.007 ± 0.001 us/op
+ * AccumulatorBenchmark.longAdderGroupAccumulateAnd_highContention avgt 15 4.770 ± 1.795 us/op
+ * AccumulatorBenchmark.longAdderGroupAccumulateAnd_lowContention avgt 15 0.061 ± 0.007 us/op
+ * AccumulatorBenchmark.longAdderGroupAccumulateAnd8_highContention avgt 15 6.025 ± 0.712 us/op
+ * AccumulatorBenchmark.longAdderGroupAccumulateAnd8_lowContention avgt 15 0.085 ± 0.004 us/op
+ * AccumulatorBenchmark.longAdderGroupIncrement_highContention avgt 15 2.294 ± 0.101 us/op
+ * AccumulatorBenchmark.longAdderGroupIncrement_lowContention avgt 15 0.019 ± 0.001 us/op
+ * AccumulatorBenchmark.longAdderGroupIncrement8_highContention avgt 15 0.786 ± 0.078 us/op
+ * AccumulatorBenchmark.longAdderGroupIncrement8_lowContention avgt 15 0.020 ± 0.001 us/op
+ * On the write side, {@link Accumulator} beats {@code longAdderGroup} at high contention by
+ * ~255x in the degenerate single-shared-lock case and still by ~46x once counters are fairly spread
+ * across 8 locks -- a large, reproducible win either way, on the call that runs on every event.
+ * On the drain side, the two designs are close and the comparison is noisy under contention for
+ * both: at width 1 {@link Accumulator}'s drain (2.746 us/op) is actually faster than {@code
+ * longAdderGroup}'s (4.770 ± 1.795 us/op, itself high-variance), and at width 8 it's only ~1.14x
+ * slower (6.875 vs 6.025 us/op) -- not the regression an earlier reading of this benchmark
+ * suggested. That earlier reading (13.357 us/op at Fork(2)) turned out to be a correlated anomaly
+ * across two independent low-sample runs, not a reproducible result; escalating to Fork(5) (15
+ * samples) settled it. Net: a large, robust win on the call that fires on every event, and no
+ * confirmed cost on the call that fires once per reporting cycle.
*/
@State(Scope.Benchmark)
@Warmup(iterations = 1, time = 10)
@Measurement(iterations = 3, time = 10)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(MICROSECONDS)
-@Fork(2)
+@Fork(5)
public class AccumulatorBenchmark {
enum Counter {
HITS
}
+ /**
+ * An 8-constant counterpart to {@link Counter}, used only by the {@code *8_*} benchmarks below.
+ * Unlike {@link Counter}, where every thread hits the single {@code HITS} constant (the worst
+ * case for {@code longAdderGroup}'s per-counter locking -- one lock shared by every thread,
+ * regardless of core count), these benchmarks spread writes across all 8 constants: each JMH
+ * worker thread is pinned to one fixed counter for its lifetime (see {@link #threadCounterIndex}),
+ * so under high contention, threads split into up to 8 groups each contending on their own lock
+ * instead of all threads sharing one. This is the topology where {@code longAdderGroup}'s
+ * distributed locking should actually pay off, and where {@link Accumulator}'s thread-striped
+ * design has to earn its win on the write side rather than facing a single-counter worst case.
+ * {@code accumulateAndReset}/{@code groupAccumulateAnd} also now walk 8 slots per drain instead of
+ * 1, sizing the drain cost closer to {@code TracerHealthMetric}'s 54-constant production shape.
+ */
+ enum Counter8 {
+ COUNTER_0,
+ COUNTER_1,
+ COUNTER_2,
+ COUNTER_3,
+ COUNTER_4,
+ COUNTER_5,
+ COUNTER_6,
+ COUNTER_7
+ }
+
+ private static final Counter8[] COUNTER8_VALUES = Counter8.values();
+
private final LongAdder adder = new LongAdder();
- private final long[][] accumulator = Accumulator.EmbeddingSupport.create(Counter.values());
- private final Accumulator typedAccumulator = Accumulator.of(Counter.values());
+ private final Accumulator accumulator = Accumulator.of(Counter.values());
+ private final Accumulator accumulator8 = Accumulator.of(Counter8.values());
private final ConcurrentHashMap chm = new ConcurrentHashMap<>();
private final LongAdder[] longAdderGroup = {new LongAdder()};
+ private final LongAdder[] longAdderGroup8 = {
+ new LongAdder(),
+ new LongAdder(),
+ new LongAdder(),
+ new LongAdder(),
+ new LongAdder(),
+ new LongAdder(),
+ new LongAdder(),
+ new LongAdder()
+ };
+
+ /**
+ * Assigns each JMH worker thread a fixed {@code Counter8} index (round-robin over 8) the first
+ * time it calls into any {@code *8_*} benchmark, and keeps returning that same index for the
+ * thread's lifetime -- so under {@code Threads.MAX}, writes spread across all 8 counters instead
+ * of every thread hammering one.
+ */
+ private final AtomicInteger threadIndexAssigner = new AtomicInteger();
+
+ private final ThreadLocal threadCounterIndex =
+ ThreadLocal.withInitial(() -> threadIndexAssigner.getAndIncrement() % COUNTER8_VALUES.length);
/**
* The natural "just use LongAdder" fix for the reset hazard: one {@code LongAdder} per counter,
* with a per-counter lock guarding both the increment and the drain -- external locking around
* only the drain does nothing, since {@code sumThenReset()}'s internal race is against the {@code
* LongAdder}'s own CAS-based {@code add()}, not against any lock a caller takes. This is the fair
- * comparison point: it closes the same hazard {@link Accumulator} does, but stripes by
- * counter (one lock per enum constant) instead of by thread (one lock per
- * stripe, shared by all counters) -- so N threads hammering the *same* counter contend on one
- * lock regardless of core count, with no thread-bucket distribution at all.
+ * comparison point: it closes the same reset hazard {@link Accumulator} does, but stripes by
+ * counter (one lock per enum constant) instead of by thread (one shared table
+ * across all counters) -- so N threads hammering the *same* counter contend on one lock
+ * regardless of core count, with no thread-bucket distribution at all.
*/
private static void groupInc(LongAdder[] group, int ordinal) {
LongAdder counter = group[ordinal];
@@ -176,31 +184,13 @@ public void longAdderIncrement_highContention() {
@Benchmark
@Threads(1)
public void accumulatorIncrement_lowContention() {
- Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS);
+ accumulator.inc(Counter.HITS);
}
@Benchmark
@Threads(Threads.MAX)
public void accumulatorIncrement_highContention() {
- Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS);
- }
-
- /**
- * The typed {@link Accumulator} wrapper's {@link Accumulator#inc}, paired against {@code
- * accumulatorIncrement*} above: same underlying {@link Accumulator.EmbeddingSupport#inc} call,
- * one extra field-load indirection through the instance. Should track the raw numbers closely --
- * a divergence here would mean the indirection isn't being inlined away.
- */
- @Benchmark
- @Threads(1)
- public void typedIncrement_lowContention() {
- typedAccumulator.inc(Counter.HITS);
- }
-
- @Benchmark
- @Threads(Threads.MAX)
- public void typedIncrement_highContention() {
- typedAccumulator.inc(Counter.HITS);
+ accumulator.inc(Counter.HITS);
}
@Benchmark
@@ -232,9 +222,8 @@ public void longAdderSumThenReset_highContention(Blackhole blackhole) {
@Benchmark
@Threads(1)
public void accumulatorAccumulateAndReset_lowContention(Blackhole blackhole) {
- Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS);
- blackhole.consume(
- Accumulator.EmbeddingSupport.accumulateAndReset(accumulator, Counter.values().length));
+ accumulator.inc(Counter.HITS);
+ blackhole.consume(accumulator.accumulateAndReset());
}
/**
@@ -242,105 +231,113 @@ public void accumulatorAccumulateAndReset_lowContention(Blackhole blackhole) {
* Threads.MAX} threads are all draining concurrently. Real callers don't do this -- see {@code
* accumulatorMixed-write}/{@code accumulatorMixed-drain} below for the "many writers, one rare
* drainer" shape this class actually targets. Kept as the worst-case upper bound: no production
- * topology should be more contended on {@link Accumulator.EmbeddingSupport#accumulateAndReset}
- * than this.
+ * topology should be more contended on {@link Accumulator#accumulateAndReset} than this.
*/
@Benchmark
@Threads(Threads.MAX)
public void accumulatorAccumulateAndReset_highContention(Blackhole blackhole) {
- Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS);
- blackhole.consume(
- Accumulator.EmbeddingSupport.accumulateAndReset(accumulator, Counter.values().length));
+ accumulator.inc(Counter.HITS);
+ blackhole.consume(accumulator.accumulateAndReset());
}
/**
* The realistic counterpart to {@code accumulatorAccumulateAndReset_highContention}: many writer
* threads incrementing, and a single dedicated thread polling {@link
- * Accumulator.EmbeddingSupport#accumulateAndReset} -- not every thread doing both on every op.
- * {@code accumulatorMixed-write} measures increment cost while a drain is actively contending for
- * stripe locks; {@code accumulatorMixed-drain} measures the drain's own cost under that same live
- * write pressure. The 4:1 writer:drainer ratio is illustrative of "many writers, rare drain," not
- * tuned to a specific core count.
+ * Accumulator#accumulateAndReset} -- not every thread doing both on every op. {@code
+ * accumulatorMixed-write} measures increment cost while a drain is actively running; {@code
+ * accumulatorMixed-drain} measures the drain's own cost under that same live write pressure. The
+ * 4:1 writer:drainer ratio is illustrative of "many writers, rare drain," not tuned to a specific
+ * core count.
*/
@Benchmark
@Group("accumulatorMixed")
@GroupThreads(4)
public void accumulatorMixed_write() {
- Accumulator.EmbeddingSupport.inc(accumulator, Counter.HITS);
+ accumulator.inc(Counter.HITS);
}
@Benchmark
@Group("accumulatorMixed")
@GroupThreads(1)
public void accumulatorMixed_drain(Blackhole blackhole) {
- blackhole.consume(
- Accumulator.EmbeddingSupport.accumulateAndReset(accumulator, Counter.values().length));
+ blackhole.consume(accumulator.accumulateAndReset());
}
- /**
- * {@link Accumulator#update}, paired against the raw {@link Accumulator.EmbeddingSupport#update}
- * lock/dispatch it wraps: the mutator here constructs a {@link Accumulator.Stripe} under the held
- * lock and immediately lets it go, which is exactly the "small, non-capturing, non-escaping"
- * shape documented as a scalar-replacement candidate. If escape analysis is doing its job, this
- * tracks the raw call closely; if it regresses (e.g. after a JIT/JDK change, or a mutator shape
- * that stops inlining), this is the number that would move.
- */
@Benchmark
@Threads(1)
- public void typedUpdate_lowContention() {
- typedAccumulator.update(stripe -> stripe.inc(Counter.HITS));
+ public void longAdderGroupIncrement_lowContention() {
+ groupInc(longAdderGroup, Counter.HITS.ordinal());
}
@Benchmark
@Threads(Threads.MAX)
- public void typedUpdate_highContention() {
- typedAccumulator.update(stripe -> stripe.inc(Counter.HITS));
+ public void longAdderGroupIncrement_highContention() {
+ groupInc(longAdderGroup, Counter.HITS.ordinal());
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void longAdderGroupAccumulateAnd_lowContention(Blackhole blackhole) {
+ groupInc(longAdderGroup, Counter.HITS.ordinal());
+ blackhole.consume(groupAccumulateAnd(longAdderGroup));
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void longAdderGroupAccumulateAnd_highContention(Blackhole blackhole) {
+ groupInc(longAdderGroup, Counter.HITS.ordinal());
+ blackhole.consume(groupAccumulateAnd(longAdderGroup));
}
- /**
- * {@link Accumulator#accumulateAndReset}, paired against the raw {@link
- * Accumulator.EmbeddingSupport#accumulateAndReset} it wraps. Unlike {@link Accumulator.Stripe},
- * {@link Accumulator.Counts} is documented to escape (the caller holds and reads it after
- * return), so this is expected to run measurably slower than the raw call by roughly one small
- * object allocation per drain -- not a scalar-replacement candidate, and not meant to look free.
- */
@Benchmark
@Threads(1)
- public void typedAccumulateAndReset_lowContention(Blackhole blackhole) {
- typedAccumulator.inc(Counter.HITS);
- blackhole.consume(typedAccumulator.accumulateAndReset());
+ public void accumulatorIncrement8_lowContention() {
+ accumulator8.inc(COUNTER8_VALUES[threadCounterIndex.get()]);
}
@Benchmark
@Threads(Threads.MAX)
- public void typedAccumulateAndReset_highContention(Blackhole blackhole) {
- typedAccumulator.inc(Counter.HITS);
- blackhole.consume(typedAccumulator.accumulateAndReset());
+ public void accumulatorIncrement8_highContention() {
+ accumulator8.inc(COUNTER8_VALUES[threadCounterIndex.get()]);
}
@Benchmark
@Threads(1)
- public void longAdderGroupIncrement_lowContention() {
- groupInc(longAdderGroup, Counter.HITS.ordinal());
+ public void accumulatorAccumulateAndReset8_lowContention(Blackhole blackhole) {
+ accumulator8.inc(COUNTER8_VALUES[threadCounterIndex.get()]);
+ blackhole.consume(accumulator8.accumulateAndReset());
}
@Benchmark
@Threads(Threads.MAX)
- public void longAdderGroupIncrement_highContention() {
- groupInc(longAdderGroup, Counter.HITS.ordinal());
+ public void accumulatorAccumulateAndReset8_highContention(Blackhole blackhole) {
+ accumulator8.inc(COUNTER8_VALUES[threadCounterIndex.get()]);
+ blackhole.consume(accumulator8.accumulateAndReset());
}
@Benchmark
@Threads(1)
- public void longAdderGroupAccumulateAnd_lowContention(Blackhole blackhole) {
- groupInc(longAdderGroup, Counter.HITS.ordinal());
- blackhole.consume(groupAccumulateAnd(longAdderGroup));
+ public void longAdderGroupIncrement8_lowContention() {
+ groupInc(longAdderGroup8, threadCounterIndex.get());
}
@Benchmark
@Threads(Threads.MAX)
- public void longAdderGroupAccumulateAnd_highContention(Blackhole blackhole) {
- groupInc(longAdderGroup, Counter.HITS.ordinal());
- blackhole.consume(groupAccumulateAnd(longAdderGroup));
+ public void longAdderGroupIncrement8_highContention() {
+ groupInc(longAdderGroup8, threadCounterIndex.get());
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void longAdderGroupAccumulateAnd8_lowContention(Blackhole blackhole) {
+ groupInc(longAdderGroup8, threadCounterIndex.get());
+ blackhole.consume(groupAccumulateAnd(longAdderGroup8));
+ }
+
+ @Benchmark
+ @Threads(Threads.MAX)
+ public void longAdderGroupAccumulateAnd8_highContention(Blackhole blackhole) {
+ groupInc(longAdderGroup8, threadCounterIndex.get());
+ blackhole.consume(groupAccumulateAnd(longAdderGroup8));
}
}
diff --git a/internal-api/src/main/java/datadog/trace/util/Accumulator.java b/internal-api/src/main/java/datadog/trace/util/Accumulator.java
index 746c985a5b6..bb830b4acce 100644
--- a/internal-api/src/main/java/datadog/trace/util/Accumulator.java
+++ b/internal-api/src/main/java/datadog/trace/util/Accumulator.java
@@ -1,44 +1,47 @@
package datadog.trace.util;
import datadog.environment.ThreadSupport;
-import datadog.trace.api.function.Strategy;
-import datadog.trace.api.function.StrategyConsumer;
-import java.util.Arrays;
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
-import java.util.function.ObjLongConsumer;
-import javax.annotation.ParametersAreNonnullByDefault;
-import javax.annotation.concurrent.GuardedBy;
+import java.util.concurrent.atomic.AtomicLongArray;
/**
- * A typed, instance-owning wrapper over {@link EmbeddingSupport}: ties an enum's type to its
- * backing {@code long[][]} at construction, so {@link #inc}/{@link #add} can't be called with a key
- * from a different enum than the one this accumulator was {@link #of created} for. Costs one
- * field-load indirection per call versus calling {@link EmbeddingSupport} directly -- the same
- * trade {@code StringIndex} makes over its own nested {@code EmbeddingSupport}.
+ * A striped, lock-free counter primitive keyed by enum ordinal: {@code LongAdder}'s write
+ * scalability, but as one shared, thread-sharded table instead of one independent {@code LongAdder}
+ * per counter -- which avoids paying {@code LongAdder}'s per-instance striping overhead {@code
+ * E.values().length} times over.
*
* {@code
* enum MyCounters { FOO, BAR }
*
* Accumulator counters = Accumulator.of(MyCounters.values());
* counters.inc(MyCounters.FOO);
- * counters.update(stripe -> {
- * stripe.inc(MyCounters.FOO);
- * stripe.inc(MyCounters.BAR);
- * });
+ * counters.add(MyCounters.BAR, 5L);
*
- * Accumulator.Counts drained = counters.accumulateAndReset(); // atomically per stripe
+ * Accumulator.Counts drained = counters.accumulateAndReset();
* long foo = drained.get(MyCounters.FOO);
* }
*
- * @see EmbeddingSupport
+ * Each counter's own {@link #accumulateAndReset} slot is read-and-zeroed with a single atomic
+ * {@code getAndSet}, so -- like {@code Accumulator}'s previous {@code synchronized}-stripe design,
+ * and unlike {@code LongAdder#sumThenReset()} -- no individual increment can land in the gap
+ * between summing and zeroing and be silently lost. What's gone is the previous design's
+ * row-wide atomicity: {@link #inc}/{@link #add} for two different counters are no longer
+ * guaranteed to be seen together by a concurrent {@link #accumulateAndReset}. There is no {@code
+ * update}-style escape hatch for grouping several counters under one atomic operation -- callers
+ * needing that must weigh whether the guarantee was load-bearing (most call sites are logging
+ * unrelated aspects of the same event, not maintaining a cross-counter invariant a reader depends
+ * on) or bring their own coordination.
*/
public final class Accumulator> {
- private final long[][] data;
+ /** One full cache line of {@code long}s (64 bytes), used to pad each stripe row. */
+ private static final int CACHE_LINE_LONGS = 8;
+
+ private final AtomicLongArray[] data;
+ private final int width;
private final E[] values;
- private Accumulator(long[][] data, E[] values) {
+ private Accumulator(AtomicLongArray[] data, int width, E[] values) {
this.data = data;
+ this.width = width;
this.values = values;
}
@@ -46,7 +49,14 @@ private Accumulator(long[][] data, E[] values) {
* @param values the enum constants naming each counter, e.g. {@code MyCounters.values()}
*/
public static > Accumulator of(E[] values) {
- return new Accumulator<>(EmbeddingSupport.create(values), values);
+ int width = values.length;
+ int paddedWidth = paddedWidth(width);
+ int stripes = stripeCount();
+ AtomicLongArray[] data = new AtomicLongArray[stripes];
+ for (int i = 0; i < stripes; i++) {
+ data[i] = new AtomicLongArray(paddedWidth);
+ }
+ return new Accumulator<>(data, width, values);
}
/**
@@ -58,118 +68,29 @@ public static > Accumulator of(Class enumType) {
/** Increments the counter named by {@code key} in the calling thread's stripe by one. */
public void inc(E key) {
- EmbeddingSupport.inc(data, key);
+ add(key, 1L);
}
/** Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe. */
public void add(E key, long delta) {
- EmbeddingSupport.add(data, key, delta);
- }
-
- /**
- * Runs {@code mutator} against a typed view of the calling thread's stripe under a single held
- * lock -- the escape hatch for performing several related updates atomically with respect to a
- * concurrent {@link #accumulateAndReset}.
- *
- * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it
- * inlines into the lock's critical section, and don't let the {@link Stripe} escape it (store
- * it, return it, hand it to another thread) -- see {@link Stripe}
- */
- @StrategyConsumer
- public void update(@Strategy Consumer> mutator) {
- long[] stripe = EmbeddingSupport.stripeOf(data);
- synchronized (stripe) {
- mutator.accept(new Stripe<>(stripe));
- }
- }
-
- /**
- * Like {@link #update(Consumer)}, but passes {@code context} to {@code mutator} as an explicit
- * parameter instead of letting the mutator capture it -- for a caller that would otherwise need
- * to close over a local (e.g. a count) just to get it into the critical section. Note {@code
- * context} is boxed if it's a primitive at the call site; that's a real allocation trade against
- * the capturing lambda it replaces, not a free win -- prefer this only when {@code context} would
- * otherwise be the only thing forcing a capture. For an {@code int} or {@code long} context, use
- * {@link #update(long, ObjLongConsumer)} instead to avoid that boxing entirely.
- *
- * @param context a value the mutator needs, passed in rather than captured
- * @param mutator a strategy over {@code context} and the selected stripe; keep it small and
- * non-capturing so it inlines into the lock's critical section, and don't let the {@link
- * Stripe} escape it (store it, return it, hand it to another thread) -- see {@link Stripe}
- */
- @StrategyConsumer
- public void update(C context, @Strategy BiConsumer> mutator) {
- long[] stripe = EmbeddingSupport.stripeOf(data);
- synchronized (stripe) {
- mutator.accept(context, new Stripe<>(stripe));
- }
- }
-
- /**
- * Like {@link #update(Object, BiConsumer)}, but for a {@code long} context -- reuses the JDK's
- * {@link ObjLongConsumer} instead of the generic {@link BiConsumer}, so {@code context} is passed
- * as a primitive {@code long} rather than boxed into a {@link Long}. Covers an {@code int}
- * context too: it widens to {@code long} for free at the call site, no boxing either way. (A
- * dedicated {@code int} overload isn't offered alongside this one -- an {@code int} argument
- * would be ambiguous between the two, since it's an exact match for one and a free widening
- * conversion to the other, and unrelated functional-interface types block the usual most-specific
- * tiebreak.)
- *
- * Note the parameter order this forces: {@link ObjLongConsumer#accept} takes {@code (T,
- * long)}, so the mutator sees the stripe first and the context second -- the opposite order from
- * {@link #update(Object, BiConsumer)}.
- *
- * @param context a primitive value the mutator needs, passed in rather than captured or boxed
- * @param mutator a strategy over the selected stripe and {@code context}; keep it small and
- * non-capturing so it inlines into the lock's critical section, and don't let the {@link
- * Stripe} escape it (store it, return it, hand it to another thread) -- see {@link Stripe}
- */
- @StrategyConsumer
- public void update(long context, @Strategy ObjLongConsumer> mutator) {
- long[] stripe = EmbeddingSupport.stripeOf(data);
- synchronized (stripe) {
- mutator.accept(new Stripe<>(stripe), context);
- }
+ stripeOf(data).getAndAdd(key.ordinal(), delta);
}
/**
- * A typed view over one stripe, handed to an {@link #update} strategy: the same enum-ordinal type
- * checking {@link Accumulator} provides at the top level, applied inside the critical section
- * too.
- *
- * Constructed fresh under the held lock on every {@link #update} call. A well-behaved {@link
- * Strategy} mutator -- small, non-capturing, and never storing or returning this object -- lets
- * escape analysis prove it doesn't escape the inlined call and scalar-replace it, so no
- * allocation survives to run time. Break those rules (capture it in a field, return it, hand it
- * to another thread) and it degrades to a real, per-call allocation instead of a compile-time
- * fiction with no correctness difference either way -- just a cost one.
- */
- public static final class Stripe> {
- private final long[] stripe;
-
- private Stripe(long[] stripe) {
- this.stripe = stripe;
- }
-
- /** Increments the counter named by {@code key} in this stripe by one. */
- public void inc(E key) {
- EmbeddingSupport.inc(stripe, key);
- }
-
- /** Adds {@code delta} to the counter named by {@code key} in this stripe. */
- public void add(E key, long delta) {
- EmbeddingSupport.add(stripe, key, delta);
- }
- }
-
- /**
- * Combines and resets every stripe, returning the sum as a typed view.
+ * Combines and resets every stripe, returning the sum as a typed view. Each counter is
+ * read-and-zeroed with one atomic {@code getAndSet} -- see the class-level note on what atomicity
+ * this does and doesn't provide across different counters.
*
* @return the sum, keyed by the enum's {@code ordinal()}
- * @see EmbeddingSupport#accumulateAndReset
*/
public Counts accumulateAndReset() {
- return new Counts<>(EmbeddingSupport.accumulateAndReset(data, values.length), values);
+ long[] acc = new long[width];
+ for (AtomicLongArray stripe : data) {
+ for (int i = 0; i < width; i++) {
+ acc[i] += stripe.getAndSet(i, 0L);
+ }
+ }
+ return new Counts<>(acc, values);
}
/**
@@ -178,21 +99,21 @@ public Counts accumulateAndReset() {
* the delta a concurrent {@link #accumulateAndReset} on a reporting cadence is about to report.
*
* @return the sum, keyed by the enum's {@code ordinal()}
- * @see EmbeddingSupport#sum(long[][], int)
*/
public Counts sum() {
- return new Counts<>(EmbeddingSupport.sum(data, values.length), values);
+ long[] acc = new long[width];
+ for (AtomicLongArray stripe : data) {
+ for (int i = 0; i < width; i++) {
+ acc[i] += stripe.get(i);
+ }
+ }
+ return new Counts<>(acc, values);
}
/**
- * A typed view over a drained {@code long[]}, returned by {@link #accumulateAndReset} or {@link
- * #sum}: the same enum-ordinal type checking {@link Accumulator} provides on writes, applied to
- * the read side too.
- *
- * Unlike {@link Stripe}, this is expected to escape -- the caller holds and reads it after the
- * call returns -- so it's a real, per-drain allocation, not a scalar-replacement candidate.
- * That's fine: {@link #accumulateAndReset} runs on a reporting cadence, not per {@link
- * #inc}/{@link #add} call.
+ * A typed view over a drained snapshot, returned by {@link #accumulateAndReset} or {@link #sum}:
+ * the same enum-ordinal type checking {@link Accumulator} provides on writes, applied to the read
+ * side too.
*/
public static final class Counts> {
private final long[] counts;
@@ -251,232 +172,39 @@ public Counts plus(Counts other) {
}
/**
- * The static, raw-array tier of the striped accumulator primitive: {@code LongAdder}'s write
- * scalability, without {@code LongAdder}'s reset hazard.
+ * The calling thread's stripe: cheap masking, no allocation, no map lookup.
*
- * {@code LongAdder#sumThenReset()} is documented as not atomic against concurrent
- * updates: an increment landing on a cell after it's summed but before it's zeroed is silently
- * and permanently lost. {@link #accumulateAndReset} closes that window by combining and resetting
- * each stripe under the same lock that guards its writers.
- *
- *
Each stripe's state is a bare {@code long[]}, not a named-field struct. An {@code enum}
- * assigns a name to each position via its ordinal, so name and position are the same declaration
- * and cannot drift apart. This also makes {@link #combine} and {@link #reset} generic,
- * branchless, fixed-trip-count array loops -- the shape designed to take advantage of SIMD /
- * vector operations on modern hardware -- so they are implemented once here instead of once per
- * caller.
- *
- *
This is a pure namespace over caller-owned {@code long[][]} state -- it allocates no
- * container object and is not itself a strategy consumer's receiver. That means {@code create}'s
- * type parameter is not bound to the one later {@code inc}/{@code add} calls infer: nothing stops
- * a caller from indexing the same {@code long[][]} with a different enum than the one it was
- * {@link #create}d for, which silently reads/writes the wrong slot rather than failing to
- * compile. Prefer the owning {@link Accumulator} instance, which closes that hole for one
- * field-load indirection per call; reach for this class directly only when that indirection is
- * worth removing.
- *
- *
{@code
- * enum MyCounters { FOO, BAR }
- *
- * long[][] data = Accumulator.EmbeddingSupport.create(MyCounters.values());
- * Accumulator.EmbeddingSupport.inc(data, MyCounters.FOO);
- * Accumulator.EmbeddingSupport.update(data, stripe -> {
- * Accumulator.EmbeddingSupport.inc(stripe, MyCounters.FOO);
- * Accumulator.EmbeddingSupport.inc(stripe, MyCounters.BAR);
- * });
- *
- * long[] drained =
- * Accumulator.EmbeddingSupport.accumulateAndReset(data, MyCounters.values().length);
- * long foo = drained[MyCounters.FOO.ordinal()];
- * }
+ * Multiple threads can map to the same stripe (this is masking, not a bijection); each
+ * counter's own atomic slot makes that safe, just not maximally scalable under a hash collision.
*/
- @ParametersAreNonnullByDefault
- public static final class EmbeddingSupport {
- private EmbeddingSupport() {}
-
- /** One full cache line of {@code long}s (64 bytes), used to pad each stripe's row. */
- private static final int CACHE_LINE_LONGS = 8;
-
- /**
- * Creates the backing storage for an accumulator over {@code values}: one {@code long[]} row
- * per stripe, sized to {@code values.length} plus at least one trailing cache line of padding
- * so adjacent stripe rows don't false-share.
- *
- *
Stripe count is fixed at a power of two oversized to roughly 2x {@link
- * Runtime#availableProcessors()} (minimum 4); it is not a per-call knob (see {@link
- * #stripeCount()}).
- *
- * @param values the enum constants naming each counter, e.g. {@code MyCounters.values()}
- * @return a new {@code long[stripeCount][paddedWidth]} array, zero-initialized
- */
- public static > long[][] create(E[] values) {
- int paddedWidth = paddedWidth(values.length);
- int stripes = stripeCount();
- long[][] data = new long[stripes][];
- for (int i = 0; i < stripes; i++) {
- data[i] = new long[paddedWidth];
- }
- return data;
- }
-
- /**
- * Increments the counter named by {@code key} in the calling thread's stripe by one.
- *
- * Convenience for the common case: selects the calling thread's stripe, takes its lock, and
- * increments. To perform several increments under a single held lock, use {@link #update}.
- */
- public static > void inc(long[][] data, E key) {
- add(data, key, 1L);
- }
-
- /**
- * Adds {@code delta} to the counter named by {@code key} in the calling thread's stripe.
- *
- * @see #inc(long[][], Enum)
- */
- public static > void add(long[][] data, E key, long delta) {
- add(stripeOf(data), key, delta);
- }
-
- /**
- * Increments the counter named by {@code key} in {@code stripe} by one, under {@code stripe}'s
- * own lock.
- *
- * Intended for use inside an {@link #update} lambda, where {@code stripe} is already the
- * calling thread's selected row: {@code synchronized} is reentrant, so calling this here does
- * not deadlock or take a second lock.
- */
- public static > void inc(long[] stripe, E key) {
- add(stripe, key, 1L);
- }
-
- /**
- * Adds {@code delta} to the counter named by {@code key} in {@code stripe}, under {@code
- * stripe}'s own lock.
- *
- * @see #inc(long[], Enum)
- */
- public static > void add(long[] stripe, E key, long delta) {
- synchronized (stripe) {
- stripe[key.ordinal()] += delta;
- }
- }
-
- /**
- * Runs {@code mutator} against the calling thread's stripe under a single held lock -- the
- * escape hatch for performing several related updates atomically with respect to a concurrent
- * {@link #accumulateAndReset}.
- *
- * @param mutator a strategy over the selected stripe; keep it small and non-capturing so it
- * inlines into the lock's critical section
- */
- @StrategyConsumer
- public static void update(long[][] data, @Strategy Consumer mutator) {
- long[] stripe = stripeOf(data);
- synchronized (stripe) {
- mutator.accept(stripe);
- }
- }
-
- /**
- * Combines and resets every stripe, returning the sum. Each stripe is locked for exactly as
- * long as it takes to fold its values into the result and zero it -- the same lock held by
- * {@link #inc}/{@link #add}/{@link #update} -- so no writer can land an increment in the gap
- * between summing and zeroing the way {@code LongAdder#sumThenReset()} allows.
- *
- * Only the first {@code width} positions of each stripe are read or written -- the trailing
- * cache line {@link #paddedWidth} reserves past that point is never touched again after {@link
- * #create} zero-initializes it, so it stays a genuinely dead buffer between adjacent stripe
- * rows instead of being read-and-rewritten (dirtying that cache line) on every drain.
- *
- * @param width the number of counters actually in use, e.g. {@code values.length}
- * @return a new array of length {@code width}, indexed by the enum's {@code ordinal()}
- */
- public static long[] accumulateAndReset(long[][] data, int width) {
- long[] acc = new long[width];
- for (long[] stripe : data) {
- synchronized (stripe) {
- combine(acc, stripe, width);
- reset(stripe, width);
- }
- }
- return acc;
- }
-
- /**
- * Combines every stripe without resetting it, returning the sum -- a live, non-destructive
- * snapshot for a diagnostic read that must not perturb the delta a concurrent {@link
- * #accumulateAndReset} on a reporting cadence is about to report.
- *
- * @param width the number of counters actually in use, e.g. {@code values.length}
- * @return a new array of length {@code width}, indexed by the enum's {@code ordinal()}
- * @see #accumulateAndReset(long[][], int)
- */
- public static long[] sum(long[][] data, int width) {
- long[] acc = new long[width];
- for (long[] stripe : data) {
- synchronized (stripe) {
- combine(acc, stripe, width);
- }
- }
- return acc;
- }
-
- /**
- * {@code acc[i] += stripe[i]} for {@code i} in {@code [0, width)} -- a fixed-trip-count loop C2
- * can auto-vectorize.
- */
- @GuardedBy("stripe")
- private static void combine(long[] acc, long[] stripe, int width) {
- for (int i = 0; i < width; i++) {
- acc[i] += stripe[i];
- }
- }
-
- /**
- * Zeroes {@code stripe}'s first {@code width} positions, via the JVM-intrinsic {@link
- * Arrays#fill}. Deliberately stops at {@code width}, leaving the trailing padding untouched.
- */
- @GuardedBy("stripe")
- private static void reset(long[] stripe, int width) {
- Arrays.fill(stripe, 0, width, 0L);
- }
-
- /**
- * The calling thread's stripe: cheap masking, no allocation, no map lookup.
- *
- *
Multiple threads can map to the same stripe (this is masking, not a bijection); each
- * stripe's own lock makes that safe, just not maximally scalable under a hash collision.
- */
- private static long[] stripeOf(long[][] data) {
- int mask = data.length - 1;
- int idx = (int) (ThreadSupport.threadId() & mask);
- return data[idx];
- }
+ private static AtomicLongArray stripeOf(AtomicLongArray[] data) {
+ int mask = data.length - 1;
+ int idx = (int) (ThreadSupport.threadId() & mask);
+ return data[idx];
+ }
- /**
- * A fixed, power-of-two stripe count deliberately oversized to roughly 2x {@link
- * Runtime#availableProcessors()} (minimum 4). Not exposed as a per-call override: a mandatory
- * sizing knob on every caller fails the "print test" of self-explanatory API design.
- *
- *
Sizing to exactly the core count leaves stripe collisions likely under real contention
- * (birthday-paradox math: with {@code n} contending threads and {@code m} stripes, expected
- * colliding pairs are {@code n(n-1)/(2m)}) -- and a collision costs a blocking {@code
- * synchronized} wait, not a cheap CAS retry. Doubling the stripe count roughly halves that
- * collision count for a one-time, per-accumulator memory cost, at the price of a slightly more
- * expensive (but far rarer) {@link #accumulateAndReset} drain -- the right trade given {@link
- * #inc}/ {@link #add} run on every call while {@link #accumulateAndReset} runs on a reporting
- * cadence.
- */
- private static int stripeCount() {
- int cpus = Runtime.getRuntime().availableProcessors();
- return Math.max(4, 2 * Integer.highestOneBit(Math.max(1, cpus)));
- }
+ /**
+ * A fixed, power-of-two stripe count deliberately oversized to roughly 2x {@link
+ * Runtime#availableProcessors()} (minimum 4). Not exposed as a per-call override: a mandatory
+ * sizing knob on every caller fails the "print test" of self-explanatory API design.
+ *
+ *
Sizing to exactly the core count leaves stripe collisions likely under real contention
+ * (birthday-paradox math: with {@code n} contending threads and {@code m} stripes, expected
+ * colliding pairs are {@code n(n-1)/(2m)}) -- and a collision costs CAS-retry/cache-line-bounce
+ * cost, the same problem {@code LongAdder}'s own {@code Cell[]} table exists to avoid. Doubling
+ * the stripe count roughly halves that collision count for a one-time, per-accumulator memory
+ * cost, at the price of a slightly more expensive (but far rarer) {@link #accumulateAndReset}
+ * drain -- the right trade given {@link #inc}/{@link #add} run on every call while {@link
+ * #accumulateAndReset} runs on a reporting cadence.
+ */
+ private static int stripeCount() {
+ int cpus = Runtime.getRuntime().availableProcessors();
+ return Math.max(4, 2 * Integer.highestOneBit(Math.max(1, cpus)));
+ }
- /** Rounds {@code width} up to a whole number of cache lines, plus one full trailing line. */
- private static int paddedWidth(int width) {
- int wholeLines = ((width + CACHE_LINE_LONGS - 1) / CACHE_LINE_LONGS) * CACHE_LINE_LONGS;
- return wholeLines + CACHE_LINE_LONGS;
- }
+ /** Rounds {@code width} up to a whole number of cache lines, plus one full trailing line. */
+ private static int paddedWidth(int width) {
+ int wholeLines = ((width + CACHE_LINE_LONGS - 1) / CACHE_LINE_LONGS) * CACHE_LINE_LONGS;
+ return wholeLines + CACHE_LINE_LONGS;
}
}
diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java
index 868f2ab6d0d..a8f2f6f39d8 100644
--- a/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java
+++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorFootprintTest.java
@@ -27,24 +27,14 @@
* what each actually costs once the counters they represent are hit by real concurrent writers, as
* they are on the telemetry paths this class targets.
*
- *
Measured on a 10-CPU machine (JDK 1.8.0_382 Zulu), 4 counters, {@code
- * Accumulator.EmbeddingSupport.stripeCount()} = 16:
- *
- *
{@code
- * fresh: 4 LongAdders = 160 bytes, Accumulator = 2384 bytes
- * contended: 4 LongAdders = 17560 bytes, Accumulator = 2384 bytes
- * }
- *
- * Finding: fresh, {@code LongAdder} looks ~15x lighter -- but that's an artifact of never having
- * been written to concurrently. Once real contention forces each {@code LongAdder}'s {@code Cell[]}
- * table to grow (each {@code Cell} is {@code @Contended}-padded against false sharing, the same
- * problem {@link Accumulator}'s own padding solves), the four {@code LongAdder}s alone end up over
- * 7x heavier than {@code Accumulator}'s entire fixed footprint -- and {@code Accumulator} does not
- * grow further as more contention arrives within its existing stripe count, while every additional
- * concurrently-written {@code LongAdder} keeps paying this cost independently. {@code
- * Accumulator}'s up-front cost is the more predictable one: fixed at creation, independent of
- * runtime contention, and shared (one striped array) across however many counters the caller's enum
- * declares, rather than paid per counter.
+ * {@link Accumulator}'s stripe count is fixed at creation (roughly 2x {@link
+ * Runtime#availableProcessors()}, minimum 4) and does not grow further as more contention arrives
+ * within it, while every additional concurrently-written {@code LongAdder} keeps paying its own
+ * {@code Cell[]} growth cost independently. {@code Accumulator}'s up-front cost is the more
+ * predictable one: fixed at creation, independent of runtime contention, and shared (one striped
+ * table) across however many counters the caller's enum declares, rather than paid per counter. The
+ * printed numbers below vary by run/JVM -- see the assertions for the invariants that actually
+ * matter.
*/
class AccumulatorFootprintTest {
@@ -78,7 +68,7 @@ static LongAdder[] freshAdders() {
@Test
void freshFootprint() {
LongAdder[] adders = freshAdders();
- long[][] accumulator = Accumulator.EmbeddingSupport.create(Counters.values());
+ Accumulator accumulator = Accumulator.of(Counters.values());
long adderBytes = bytes((Object) adders);
long accumulatorBytes = bytes(accumulator);
@@ -133,7 +123,7 @@ void contendedFootprint() throws InterruptedException {
}
long contendedAdderBytes = bytes((Object) adders);
- long[][] accumulator = Accumulator.EmbeddingSupport.create(Counters.values());
+ Accumulator accumulator = Accumulator.of(Counters.values());
long accumulatorBytes = bytes(accumulator);
System.out.printf(
diff --git a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java
index a767b803476..f17eab823fb 100644
--- a/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java
+++ b/internal-api/src/test/java/datadog/trace/util/AccumulatorTest.java
@@ -23,82 +23,53 @@ enum Counters {
@Test
void freshAccumulatorSumsToZero() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
+ Accumulator counters = Accumulator.of(Counters.values());
+ Accumulator.Counts drained = counters.accumulateAndReset();
for (Counters c : Counters.values()) {
- assertEquals(0L, drained[c.ordinal()]);
+ assertEquals(0L, drained.get(c));
}
}
@Test
void incIncrementsByOne() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
- Accumulator.EmbeddingSupport.inc(data, Counters.BAR);
-
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals(2L, drained[Counters.FOO.ordinal()]);
- assertEquals(1L, drained[Counters.BAR.ordinal()]);
- assertEquals(0L, drained[Counters.BAZ.ordinal()]);
+ Accumulator counters = Accumulator.of(Counters.values());
+ counters.inc(Counters.FOO);
+ counters.inc(Counters.FOO);
+ counters.inc(Counters.BAR);
+
+ Accumulator.Counts drained = counters.accumulateAndReset();
+ assertEquals(2L, drained.get(Counters.FOO));
+ assertEquals(1L, drained.get(Counters.BAR));
+ assertEquals(0L, drained.get(Counters.BAZ));
}
@Test
void addAppliesArbitraryDelta() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- Accumulator.EmbeddingSupport.add(data, Counters.BAZ, 41L);
- Accumulator.EmbeddingSupport.add(data, Counters.BAZ, 1L);
-
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals(42L, drained[Counters.BAZ.ordinal()]);
- }
+ Accumulator counters = Accumulator.of(Counters.values());
+ counters.add(Counters.BAZ, 41L);
+ counters.add(Counters.BAZ, 1L);
- @Test
- void updateAppliesSeveralOpsUnderOneLock() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- Accumulator.EmbeddingSupport.update(
- data,
- stripe -> {
- Accumulator.EmbeddingSupport.inc(stripe, Counters.FOO);
- Accumulator.EmbeddingSupport.inc(stripe, Counters.FOO);
- Accumulator.EmbeddingSupport.add(stripe, Counters.BAR, 5L);
- });
-
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals(2L, drained[Counters.FOO.ordinal()]);
- assertEquals(5L, drained[Counters.BAR.ordinal()]);
+ Accumulator.Counts drained = counters.accumulateAndReset();
+ assertEquals(42L, drained.get(Counters.BAZ));
}
@Test
void accumulateAndResetsSoASecondDrainIsZero() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
+ Accumulator counters = Accumulator.of(Counters.values());
+ counters.inc(Counters.FOO);
- long[] first = Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals(1L, first[Counters.FOO.ordinal()]);
+ Accumulator.Counts first = counters.accumulateAndReset();
+ assertEquals(1L, first.get(Counters.FOO));
- long[] second = Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
+ Accumulator.Counts second = counters.accumulateAndReset();
for (Counters c : Counters.values()) {
- assertEquals(0L, second[c.ordinal()]);
+ assertEquals(0L, second.get(c));
}
}
- @Test
- void drainedArrayIsExactlyWidthLong() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals(Counters.values().length, drained.length);
- assertTrue(drained.length < data[0].length, "drained array should exclude stripe padding");
- }
-
@Test
void concurrentIncrementsAreNotLost() throws InterruptedException {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+ Accumulator counters = Accumulator.of(Counters.values());
int threadCount = 16;
int incrementsPerThread = 10_000;
@@ -112,7 +83,7 @@ void concurrentIncrementsAreNotLost() throws InterruptedException {
try {
start.await();
for (int i = 0; i < incrementsPerThread; i++) {
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
+ counters.inc(Counters.FOO);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -127,15 +98,14 @@ void concurrentIncrementsAreNotLost() throws InterruptedException {
pool.shutdown();
}
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals((long) threadCount * incrementsPerThread, drained[Counters.FOO.ordinal()]);
+ Accumulator.Counts drained = counters.accumulateAndReset();
+ assertEquals((long) threadCount * incrementsPerThread, drained.get(Counters.FOO));
}
@Test
void concurrentAccumulateAndDuringWritesNeverExceedsWritten()
throws InterruptedException, ExecutionException, TimeoutException {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
+ Accumulator counters = Accumulator.of(Counters.values());
int threadCount = 8;
int incrementsPerThread = 5_000;
@@ -149,11 +119,9 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten()
pool.submit(
() -> {
while (!stop.get()) {
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(
- data, Counters.values().length);
+ Accumulator.Counts drained = counters.accumulateAndReset();
synchronized (runningTotal) {
- runningTotal[0] += drained[Counters.FOO.ordinal()];
+ runningTotal[0] += drained.get(Counters.FOO);
}
}
});
@@ -162,7 +130,7 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten()
pool.execute(
() -> {
for (int i = 0; i < incrementsPerThread; i++) {
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
+ counters.inc(Counters.FOO);
}
done.countDown();
});
@@ -172,10 +140,9 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten()
stop.set(true);
drainer.get(30, TimeUnit.SECONDS);
- long[] finalDrain =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
+ Accumulator.Counts finalDrain = counters.accumulateAndReset();
synchronized (runningTotal) {
- runningTotal[0] += finalDrain[Counters.FOO.ordinal()];
+ runningTotal[0] += finalDrain.get(Counters.FOO);
}
assertEquals((long) threadCount * incrementsPerThread, runningTotal[0]);
@@ -186,47 +153,30 @@ void concurrentAccumulateAndDuringWritesNeverExceedsWritten()
@Test
void sumDoesNotResetStripes() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
+ Accumulator counters = Accumulator.of(Counters.values());
+ counters.inc(Counters.FOO);
- long[] first = Accumulator.EmbeddingSupport.sum(data, Counters.values().length);
- assertEquals(1L, first[Counters.FOO.ordinal()]);
+ Accumulator.Counts first = counters.sum();
+ assertEquals(1L, first.get(Counters.FOO));
// sum() didn't reset anything, so a second sum() sees the same total
- long[] second = Accumulator.EmbeddingSupport.sum(data, Counters.values().length);
- assertEquals(1L, second[Counters.FOO.ordinal()]);
+ Accumulator.Counts second = counters.sum();
+ assertEquals(1L, second.get(Counters.FOO));
// and a real drain afterwards still sees the value sum() didn't consume
- long[] drained =
- Accumulator.EmbeddingSupport.accumulateAndReset(data, Counters.values().length);
- assertEquals(1L, drained[Counters.FOO.ordinal()]);
+ Accumulator.Counts drained = counters.accumulateAndReset();
+ assertEquals(1L, drained.get(Counters.FOO));
}
@Test
void sumReflectsIncrementsMadeAfterAnEarlierSum() {
- long[][] data = Accumulator.EmbeddingSupport.create(Counters.values());
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
- Accumulator.EmbeddingSupport.sum(data, Counters.values().length);
-
- Accumulator.EmbeddingSupport.inc(data, Counters.FOO);
- long[] second = Accumulator.EmbeddingSupport.sum(data, Counters.values().length);
- assertEquals(2L, second[Counters.FOO.ordinal()]);
- }
-
- @Test
- void typedWrapperSumDoesNotReset() {
Accumulator counters = Accumulator.of(Counters.values());
counters.inc(Counters.FOO);
- counters.add(Counters.BAR, 5L);
+ counters.sum();
- Accumulator.Counts sum = counters.sum();
- assertEquals(1L, sum.get(Counters.FOO));
- assertEquals(5L, sum.get(Counters.BAR));
-
- // still there for the real drain
- Accumulator.Counts drained = counters.accumulateAndReset();
- assertEquals(1L, drained.get(Counters.FOO));
- assertEquals(5L, drained.get(Counters.BAR));
+ counters.inc(Counters.FOO);
+ Accumulator.Counts second = counters.sum();
+ assertEquals(2L, second.get(Counters.FOO));
}
@Test
@@ -288,68 +238,4 @@ void plusCombinesAStoredRunningTotalWithALiveSumWithoutMutatingEither() {
assertEquals(1L, storedTotal.get(Counters.FOO));
assertEquals(1L, counters.sum().get(Counters.FOO));
}
-
- @Test
- void typedWrapperDelegatesToEmbeddingSupport() {
- Accumulator counters = Accumulator.of(Counters.values());
- counters.inc(Counters.FOO);
- counters.inc(Counters.FOO);
- counters.add(Counters.BAR, 5L);
- counters.update(stripe -> stripe.inc(Counters.BAZ));
-
- Accumulator.Counts drained = counters.accumulateAndReset();
- assertEquals(2L, drained.get(Counters.FOO));
- assertEquals(5L, drained.get(Counters.BAR));
- assertEquals(1L, drained.get(Counters.BAZ));
- }
-
- @Test
- void contextualUpdatePassesContextInsteadOfCapturingIt() {
- Accumulator counters = Accumulator.of(Counters.values());
- String context = "abcde";
-
- counters.update(
- context,
- (ctx, stripe) -> {
- stripe.inc(Counters.FOO);
- stripe.add(Counters.BAR, ctx.length());
- });
-
- Accumulator.Counts drained = counters.accumulateAndReset();
- assertEquals(1L, drained.get(Counters.FOO));
- assertEquals(5L, drained.get(Counters.BAR));
- }
-
- @Test
- void intContextWidensIntoTheLongOverloadWithoutBoxing() {
- Accumulator counters = Accumulator.of(Counters.values());
- int delta = 5;
-
- counters.update(
- delta,
- (stripe, d) -> {
- stripe.inc(Counters.FOO);
- stripe.add(Counters.BAR, d);
- });
-
- Accumulator.Counts drained = counters.accumulateAndReset();
- assertEquals(1L, drained.get(Counters.FOO));
- assertEquals(5L, drained.get(Counters.BAR));
- }
-
- @Test
- void longContextualUpdateAvoidsBoxing() {
- Accumulator counters = Accumulator.of(Counters.values());
-
- counters.update(
- 5L,
- (stripe, delta) -> {
- stripe.inc(Counters.FOO);
- stripe.add(Counters.BAR, delta);
- });
-
- Accumulator.Counts drained = counters.accumulateAndReset();
- assertEquals(1L, drained.get(Counters.FOO));
- assertEquals(5L, drained.get(Counters.BAR));
- }
}
diff --git a/products/metrics/metrics-api/build.gradle.kts b/products/metrics/metrics-api/build.gradle.kts
index bc995a5c87d..d092baf0a58 100644
--- a/products/metrics/metrics-api/build.gradle.kts
+++ b/products/metrics/metrics-api/build.gradle.kts
@@ -7,4 +7,7 @@ description = "Metrics API"
dependencies {
implementation(libs.slf4j)
+ implementation(project(":internal-api"))
+
+ testImplementation(libs.bundles.junit5)
}
diff --git a/products/metrics/metrics-api/src/main/java/datadog/metrics/api/statsd/StatsDCountReporter.java b/products/metrics/metrics-api/src/main/java/datadog/metrics/api/statsd/StatsDCountReporter.java
new file mode 100644
index 00000000000..089215f7caf
--- /dev/null
+++ b/products/metrics/metrics-api/src/main/java/datadog/metrics/api/statsd/StatsDCountReporter.java
@@ -0,0 +1,28 @@
+package datadog.metrics.api.statsd;
+
+import datadog.trace.util.Accumulator;
+import java.util.function.ToLongFunction;
+
+/** Reports a batch of per-key deltas to a {@link StatsDClient}, skipping unchanged keys. */
+public final class StatsDCountReporter {
+ private StatsDCountReporter() {}
+
+ public static & StatsDCounterKey> void report(
+ StatsDClient statsDClient, E[] values, ToLongFunction counts) {
+ for (E value : values) {
+ long delta = counts.applyAsLong(value);
+ if (delta != 0) {
+ statsDClient.count(value.getMetricName(), delta, value.getTags());
+ }
+ }
+ }
+
+ /**
+ * Convenience for the common case of reporting a drained {@link Accumulator.Counts} directly --
+ * the keys come along with it, so the caller doesn't need to separately pass {@code E.values()}.
+ */
+ public static & StatsDCounterKey> void report(
+ StatsDClient statsDClient, Accumulator.Counts counts) {
+ report(statsDClient, counts.keys(), counts::get);
+ }
+}
diff --git a/products/metrics/metrics-api/src/main/java/datadog/metrics/api/statsd/StatsDCounterKey.java b/products/metrics/metrics-api/src/main/java/datadog/metrics/api/statsd/StatsDCounterKey.java
new file mode 100644
index 00000000000..3aadbfc24f5
--- /dev/null
+++ b/products/metrics/metrics-api/src/main/java/datadog/metrics/api/statsd/StatsDCounterKey.java
@@ -0,0 +1,8 @@
+package datadog.metrics.api.statsd;
+
+/** A counter identity: the dogstatsd metric name and tags a batch of counts should report under. */
+public interface StatsDCounterKey {
+ String getMetricName();
+
+ String[] getTags();
+}
diff --git a/products/metrics/metrics-api/src/test/java/datadog/metrics/api/statsd/RecordingStatsDClient.java b/products/metrics/metrics-api/src/test/java/datadog/metrics/api/statsd/RecordingStatsDClient.java
new file mode 100644
index 00000000000..f017a134086
--- /dev/null
+++ b/products/metrics/metrics-api/src/test/java/datadog/metrics/api/statsd/RecordingStatsDClient.java
@@ -0,0 +1,63 @@
+package datadog.metrics.api.statsd;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Test fake that records every {@link #count} call for assertion. */
+final class RecordingStatsDClient implements StatsDClient {
+
+ static final class Count {
+ final String metricName;
+ final long delta;
+ final String[] tags;
+
+ Count(String metricName, long delta, String[] tags) {
+ this.metricName = metricName;
+ this.delta = delta;
+ this.tags = tags;
+ }
+ }
+
+ final List counts = new ArrayList<>();
+
+ @Override
+ public void incrementCounter(String metricName, String... tags) {}
+
+ @Override
+ public void count(String metricName, long delta, String... tags) {
+ counts.add(new Count(metricName, delta, tags));
+ }
+
+ @Override
+ public void gauge(String metricName, long value, String... tags) {}
+
+ @Override
+ public void gauge(String metricName, double value, String... tags) {}
+
+ @Override
+ public void histogram(String metricName, long value, String... tags) {}
+
+ @Override
+ public void histogram(String metricName, double value, String... tags) {}
+
+ @Override
+ public void distribution(String metricName, long value, String... tags) {}
+
+ @Override
+ public void distribution(String metricName, double value, String... tags) {}
+
+ @Override
+ public void serviceCheck(
+ String serviceCheckName, String status, String message, String... tags) {}
+
+ @Override
+ public void error(Exception error) {}
+
+ @Override
+ public int getErrorCount() {
+ return 0;
+ }
+
+ @Override
+ public void close() {}
+}
diff --git a/products/metrics/metrics-api/src/test/java/datadog/metrics/api/statsd/StatsDCountReporterTest.java b/products/metrics/metrics-api/src/test/java/datadog/metrics/api/statsd/StatsDCountReporterTest.java
new file mode 100644
index 00000000000..1b7263060c4
--- /dev/null
+++ b/products/metrics/metrics-api/src/test/java/datadog/metrics/api/statsd/StatsDCountReporterTest.java
@@ -0,0 +1,94 @@
+package datadog.metrics.api.statsd;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class StatsDCountReporterTest {
+
+ private static final String[] TAG_A = {"env:a"};
+ private static final String[] TAG_B = {"env:b"};
+
+ enum Counters implements StatsDCounterKey {
+ FOO("foo.total", TAG_A),
+ BAR("bar.total", TAG_A),
+ SHARED_A("shared.total", TAG_A),
+ SHARED_B("shared.total", TAG_B);
+
+ private final String metricName;
+ private final String[] tags;
+
+ Counters(String metricName, String[] tags) {
+ this.metricName = metricName;
+ this.tags = tags;
+ }
+
+ @Override
+ public String getMetricName() {
+ return metricName;
+ }
+
+ @Override
+ public String[] getTags() {
+ return tags;
+ }
+ }
+
+ @Test
+ void reportsNonZeroCounterWithItsOwnMetricNameAndTags() {
+ RecordingStatsDClient statsD = new RecordingStatsDClient();
+ Map deltas = new HashMap<>();
+ deltas.put(Counters.FOO, 3L);
+
+ StatsDCountReporter.report(statsD, Counters.values(), c -> deltas.getOrDefault(c, 0L));
+
+ assertEquals(1, statsD.counts.size());
+ RecordingStatsDClient.Count count = statsD.counts.get(0);
+ assertEquals("foo.total", count.metricName);
+ assertEquals(3L, count.delta);
+ assertArrayEquals(TAG_A, count.tags);
+ }
+
+ @Test
+ void skipsZeroDeltaCounters() {
+ RecordingStatsDClient statsD = new RecordingStatsDClient();
+
+ StatsDCountReporter.report(statsD, Counters.values(), c -> 0L);
+
+ assertTrue(statsD.counts.isEmpty());
+ }
+
+ @Test
+ void reportsNothingWhenEveryCounterIsZero() {
+ RecordingStatsDClient statsD = new RecordingStatsDClient();
+ Map deltas = new HashMap<>();
+
+ StatsDCountReporter.report(statsD, Counters.values(), c -> deltas.getOrDefault(c, 0L));
+
+ assertTrue(statsD.counts.isEmpty());
+ }
+
+ @Test
+ void reportsConstantsSharingAMetricNameIndependentlyByTag() {
+ RecordingStatsDClient statsD = new RecordingStatsDClient();
+ Map deltas = new HashMap<>();
+ deltas.put(Counters.SHARED_A, 5L);
+ deltas.put(Counters.SHARED_B, 7L);
+
+ StatsDCountReporter.report(statsD, Counters.values(), c -> deltas.getOrDefault(c, 0L));
+
+ assertEquals(2, statsD.counts.size());
+ RecordingStatsDClient.Count a = statsD.counts.get(0);
+ RecordingStatsDClient.Count b = statsD.counts.get(1);
+ assertEquals("shared.total", a.metricName);
+ assertEquals(5L, a.delta);
+ assertArrayEquals(TAG_A, a.tags);
+ assertEquals("shared.total", b.metricName);
+ assertEquals(7L, b.delta);
+ assertArrayEquals(TAG_B, b.tags);
+ }
+}