diff --git a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/AdversarialMetricsBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/AdversarialMetricsBenchmark.java index 01ba90097a4..ac5938fcc32 100644 --- a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/AdversarialMetricsBenchmark.java +++ b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/AdversarialMetricsBenchmark.java @@ -27,29 +27,35 @@ import org.openjdk.jmh.infra.Blackhole; /** - * Adversarial JMH benchmark designed to stress the metrics subsystem's capacity bounds. + * Adversarial JMH benchmark designed to stress every cardinality + capacity dimension of the + * metrics subsystem at once. * - *

The metrics aggregator is bounded at every layer: + *

The metrics aggregator is supposed to be bounded by design: * *

* - *

The benchmark hammers all of these simultaneously with 8 producer threads, unique labels per - * op (so the aggregate cache fills+evicts repeatedly), random durations across a wide range (so - * histograms accept many distinct bins), and random {@code error}/{@code topLevel} flags (so both - * histograms are exercised). After the run, drop counters are printed so you can see how the - * subsystem absorbed the burst. + *

This benchmark hammers all of those bounds simultaneously with 8 producer threads, unique + * labels per op (so handlers cap and the table fills+evicts repeatedly), random durations across a + * wide range (so histograms accept many distinct bins), and random {@code error}/{@code topLevel} + * flags (so both histograms are exercised). After the run, prints the drop counters so you can + * verify the subsystem stayed bounded under attack. * - *

What "OOM the metrics subsystem" would look like if the bounds break: producer-thread - * allocation would grow unbounded (snapshots faster than the inbox can drain produces dropped - * snapshots, not heap growth); aggregator-thread heap would grow if entries weren't capped or - * histograms grew past their dense-store limit. + *

What "OOM the metrics subsystem" looks like if the bounds break: producer-thread allocation + * would grow unbounded (snapshots faster than inbox can drain produces dropped snapshots, not heap + * growth); aggregator-thread heap would grow if entries weren't capped, if handlers grew past their + * cap, or if histograms grew past their dense-store limit. */ @State(Scope.Benchmark) @Warmup(iterations = 2, time = 15, timeUnit = SECONDS) @@ -100,7 +106,7 @@ public void tearDown() { System.err.println( " onStatsAggregateDropped = " + health.aggregateDropped.sum() - + " (snapshots dropped because the aggregate cache was full with no stale entry)"); + + " (snapshots dropped because the AggregateTable was full with no stale entry)"); } @Benchmark @@ -108,9 +114,9 @@ public void publish(ThreadState ts, Blackhole blackhole) { int idx = ts.cursor++; ThreadLocalRandom rng = ThreadLocalRandom.current(); - // Mix indices so labels don't fall into linear order. Distinct labels exceed every reasonable - // working-set bound, so the aggregate cache evicts continuously and most ops force a fresh - // MetricKey construction on the consumer thread. + // Mix indices so labels don't fall into linear order in the handler tables. Distinct labels + // exceed every cap (RESOURCE=512, OPERATION=128, SERVICE=128, peer.hostname=512), so handlers + // saturate fast and most ops resolve to the blocked-by-tracer sentinel. int scrambled = idx * 0x9E3779B1; // golden ratio multiplier String service = "svc-" + (scrambled & 0xFFFF); String operation = "op-" + ((scrambled >>> 8) & 0x3FFFF); diff --git a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorMissPathBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorMissPathBenchmark.java new file mode 100644 index 00000000000..2079cbf15ec --- /dev/null +++ b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorMissPathBenchmark.java @@ -0,0 +1,83 @@ +package datadog.trace.common.metrics; + +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND; +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT; +import static java.util.concurrent.TimeUnit.MICROSECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; + +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.trace.api.WellKnownTags; +import datadog.trace.core.CoreSpan; +import datadog.trace.core.monitor.HealthMetrics; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +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.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +/** + * Miss-path variant of {@link ClientStatsAggregatorBenchmark}. Each op publishes a single-span + * trace from a pre-built pool where every span has a unique (service, operation, resource) tuple. + * After cardinality budgets fill, fields canonicalize to the {@code blocked_by_tracer} sentinel, + * but the producer still allocates a {@link SpanSnapshot} per op and enqueues it for the aggregator + * -- so the steady state exercises the per-op publish allocations + the consumer's + * canonicalize/match work, not the hit-path-only pattern of the other benchmarks. + * + *

Run with {@code -prof gc} to compare allocation rates against master's {@code + * ConflatingMetricsAggregator}. + */ +@State(Scope.Benchmark) +@Warmup(iterations = 1, time = 15, timeUnit = SECONDS) +@Measurement(iterations = 3, time = 15, timeUnit = SECONDS) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(MICROSECONDS) +@Fork(value = 1) +public class ClientStatsAggregatorMissPathBenchmark { + + private static final int POOL_SIZE = 4096; + + private final DDAgentFeaturesDiscovery featuresDiscovery = + new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery( + Collections.singleton("peer.hostname"), Collections.emptySet()); + private final ClientStatsAggregator aggregator = + new ClientStatsAggregator( + new WellKnownTags("", "", "", "", "", ""), + Collections.emptySet(), + featuresDiscovery, + HealthMetrics.NO_OP, + new ClientStatsAggregatorBenchmark.NullSink(), + 2048, + 2048, + false); + + private final List>> pool = generatePool(POOL_SIZE); + private int cursor; + + static List>> generatePool(int n) { + List>> out = new ArrayList<>(n); + for (int i = 0; i < n; i++) { + SimpleSpan span = + new SimpleSpan( + "svc-" + i, "op-" + i, "res-" + i, "type-" + (i & 7), true, true, false, 0, 10, -1); + span.setTag(SPAN_KIND, SPAN_KIND_CLIENT); + span.setTag("peer.hostname", "host-" + i); + out.add(Collections.singletonList(span)); + } + return out; + } + + @Benchmark + public void benchmark(Blackhole blackhole) { + int idx = cursor; + cursor = (idx + 1) % POOL_SIZE; + blackhole.consume(aggregator.publish(pool.get(idx))); + } +} diff --git a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/TracePipelineBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/TracePipelineBenchmark.java new file mode 100644 index 00000000000..cc1d4a37538 --- /dev/null +++ b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/TracePipelineBenchmark.java @@ -0,0 +1,176 @@ +package datadog.trace.common.metrics; + +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND; +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT; +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_SERVER; +import static java.util.concurrent.TimeUnit.SECONDS; + +import datadog.trace.api.WellKnownTags; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.common.writer.Writer; +import datadog.trace.core.CoreTracer; +import datadog.trace.core.DDSpan; +import datadog.trace.core.monitor.HealthMetrics; +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.List; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +/** + * End-to-end JMH benchmark of a 3-span HTTP-style trace through {@link CoreTracer}: one {@code + * span.kind=server} root + two {@code span.kind=client} children, as if a service handled an + * incoming request that made two outbound HTTP calls. Children inherit the server span as parent + * via implicit scope-based parentage; the root finishes last so {@code PendingTrace.write} -> + * {@code tracer.write(trace)} -> metricsAggregator.publish + writer.write (no-op) runs + * synchronously on the producing thread. + * + *

Runs multi-threaded ({@link Threads} = 8 by default; override with {@code -t N}) so the + * allocation rate {@code -prof gc} reports reflects multiple producers hitting the shared metrics + * aggregator + writer pipeline, and so we can compare total throughput between revisions. + * + *

Reflection is used to swap the tracer's default no-op {@code metricsAggregator} for a real + * {@link ClientStatsAggregator} so the metrics pipeline actually runs. + * + *

Two modes via {@code @Param}: + * + *

+ */ +@State(Scope.Benchmark) +@Warmup(iterations = 2, time = 15, timeUnit = SECONDS) +@Measurement(iterations = 5, time = 15, timeUnit = SECONDS) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(SECONDS) +@Threads(8) +@Fork(value = 2) +public class TracePipelineBenchmark { + + @Param({"stable", "varied"}) + String mode; + + private CoreTracer tracer; + private ClientStatsAggregator aggregator; + private boolean stable; + + @State(Scope.Thread) + public static class ThreadState { + int cursor; + } + + @Setup + public void setup() throws Exception { + this.stable = "stable".equals(mode); + this.tracer = CoreTracer.builder().writer(new NoopWriter()).strictTraceWrites(false).build(); + this.aggregator = + new ClientStatsAggregator( + new WellKnownTags("", "", "", "", "", ""), + Collections.emptySet(), + new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery( + Collections.singleton("peer.hostname"), Collections.emptySet()), + HealthMetrics.NO_OP, + new ClientStatsAggregatorBenchmark.NullSink(), + 2048, + 2048, + false); + this.aggregator.start(); + // Replace the no-op aggregator the tracer was constructed with. The field is package-private + // in datadog.trace.core; reflect since this benchmark lives in the metrics package. + Field f = CoreTracer.class.getDeclaredField("metricsAggregator"); + f.setAccessible(true); + f.set(this.tracer, this.aggregator); + } + + @TearDown + public void tearDown() { + aggregator.close(); + tracer.close(); + } + + @Benchmark + public void threeSpanTrace(ThreadState ts, Blackhole blackhole) { + int idx = ts.cursor++; + String service = stable ? "svc" : "svc-" + idx; + String serverOp = stable ? "servlet.request" : "servlet.request-" + idx; + String serverResource = stable ? "GET /widgets/{id}" : "GET /widgets/" + idx; + String clientOp = stable ? "http.request" : "http.request-" + idx; + String clientResource1 = stable ? "GET /downstream-a" : "GET /downstream-a/" + idx; + String clientResource2 = stable ? "GET /downstream-b" : "GET /downstream-b/" + idx; + String hostA = stable ? "host-a" : "host-a-" + idx; + String hostB = stable ? "host-b" : "host-b-" + idx; + + AgentSpan server = tracer.startSpan("servlet", serverOp); + server.setResourceName(serverResource); + server.setServiceName(service); + server.setTag(SPAN_KIND, SPAN_KIND_SERVER); + AgentScope serverScope = tracer.activateSpan(server); + try { + AgentSpan client1 = tracer.startSpan("okhttp", clientOp); + client1.setResourceName(clientResource1); + client1.setServiceName(service); + client1.setTag(SPAN_KIND, SPAN_KIND_CLIENT); + client1.setTag("peer.hostname", hostA); + AgentScope client1Scope = tracer.activateSpan(client1); + try { + // simulated unit of in-call work would go here + } finally { + client1Scope.close(); + } + client1.finish(); + + AgentSpan client2 = tracer.startSpan("okhttp", clientOp); + client2.setResourceName(clientResource2); + client2.setServiceName(service); + client2.setTag(SPAN_KIND, SPAN_KIND_CLIENT); + client2.setTag("peer.hostname", hostB); + AgentScope client2Scope = tracer.activateSpan(client2); + try { + // simulated unit of in-call work would go here + } finally { + client2Scope.close(); + } + client2.finish(); + } finally { + serverScope.close(); + } + // Finishing the root last triggers PendingTrace.write -> tracer.write -> metrics + writer on + // this thread, since all child refs have already decremented to zero. + server.finish(); + blackhole.consume(server); + } + + private static final class NoopWriter implements Writer { + @Override + public void write(List trace) {} + + @Override + public void start() {} + + @Override + public boolean flush() { + return true; + } + + @Override + public void close() {} + + @Override + public void incrementDropCounts(int spanCount) {} + } +} diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java index ed23244ef1d..5b60f2a7781 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java @@ -5,8 +5,7 @@ import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; import datadog.trace.util.Hashtable; import datadog.trace.util.LongHashingUtils; -import java.util.ArrayList; -import java.util.Collections; +import java.util.Arrays; import java.util.List; import javax.annotation.Nullable; @@ -31,9 +30,15 @@ *

The handlers are reset on the aggregator thread every reporting cycle via {@link * #resetCardinalityHandlers()}. * + *

EMPTY-as-absent contract: all UTF8 fields are non-null. The optional fields ({@code + * serviceSource}, {@code httpMethod}, {@code httpEndpoint}, {@code grpcStatusCode}) carry {@link + * UTF8BytesString#EMPTY} when the snapshot had no value; {@link SerializingMetricWriter} tests + * against {@code EMPTY} (identity comparison on the singleton) to decide whether to emit each field + * on the wire. + * *

Deliberate cohesion. This class concentrates the per-field {@code * PropertyCardinalityHandler}/{@code TagCardinalityHandler} infrastructure, the canonicalized label - * fields, the encoded {@code peerTags} list used by the serializer, the {@link Canonical} scratch + * fields, the encoded {@code peerTags} array used by the serializer, the {@link Canonical} scratch * buffer, and the mutable counter/histogram aggregate state on a single object. The prior design * split label fields and aggregate state across separate {@code MetricKey} and {@code * AggregateMetric} instances, allocating both per unique key on miss; folding them yields one @@ -66,6 +71,9 @@ final class AggregateEntry extends Hashtable.Entry { static final long ERROR_TAG = 0x8000000000000000L; static final long TOP_LEVEL_TAG = 0x4000000000000000L; + /** Shared empty array used by entries with no peer tags. */ + private static final UTF8BytesString[] EMPTY_PEER_TAGS = new UTF8BytesString[0]; + /** * Whether cardinality limits substitute the {@code blocked_by_tracer} sentinel when a per-field * budget is exhausted. Read once at class init from {@link @@ -133,7 +141,7 @@ final class AggregateEntry extends Hashtable.Entry { final short httpStatusCode; final boolean synthetic; final boolean traceRoot; - final List peerTags; + final UTF8BytesString[] peerTags; // Mutable aggregate state -- single-thread (aggregator) writer. private final Histogram okLatencies = Histogram.newHistogram(); @@ -166,7 +174,7 @@ private AggregateEntry( short httpStatusCode, boolean synthetic, boolean traceRoot, - List peerTags) { + UTF8BytesString[] peerTags) { super(keyHash); this.resource = resource; this.service = service; @@ -204,6 +212,16 @@ AggregateEntry recordOneDuration(long tagAndDuration) { return this; } + /** Lazy-initializes {@link #errorLatencies} on the first error write. */ + private Histogram errorLatenciesForWrite() { + Histogram h = this.errorLatencies; + if (h == null) { + h = Histogram.newHistogram(); + this.errorLatencies = h; + } + return h; + } + int getErrorCount() { return errorCount; } @@ -234,16 +252,6 @@ Histogram getErrorLatencies() { return errorLatencies; } - /** Lazy-allocates {@link #errorLatencies} on the first error. */ - private Histogram errorLatenciesForWrite() { - Histogram h = errorLatencies; - if (h == null) { - h = Histogram.newHistogram(); - errorLatencies = h; - } - return h; - } - /** * Resets the per-cycle counters and histograms. Label fields ({@code resource}, {@code service}, * ..., {@code peerTags}) are deliberately left intact -- they're the entry's bucket identity and @@ -292,7 +300,10 @@ static AggregateEntry of( UTF8BytesString httpMethodUtf = createUtf8(httpMethod); UTF8BytesString httpEndpointUtf = createUtf8(httpEndpoint); UTF8BytesString grpcUtf = createUtf8(grpcStatusCode); - List peerTagsList = peerTags == null ? Collections.emptyList() : peerTags; + UTF8BytesString[] peerTagsArray = + peerTags == null || peerTags.isEmpty() + ? EMPTY_PEER_TAGS + : peerTags.toArray(new UTF8BytesString[0]); long keyHash = hashOf( resourceUtf, @@ -307,7 +318,8 @@ static AggregateEntry of( (short) httpStatusCode, synthetic, traceRoot, - peerTagsList); + peerTagsArray, + peerTagsArray.length); return new AggregateEntry( keyHash, resourceUtf, @@ -322,7 +334,7 @@ static AggregateEntry of( (short) httpStatusCode, synthetic, traceRoot, - peerTagsList); + peerTagsArray); } /** @@ -351,9 +363,13 @@ static void resetCardinalityHandlers() { * {@link #of} produce the same hash as entries built from a snapshot with matching content. * *

Field order intentionally mirrors {@link Canonical#matches} -- UTF8 fields first (highest - * cardinality first for matches' short-circuit benefit), then the peer-tag list, then the + * cardinality first for matches' short-circuit benefit), then the peer-tag array, then the * primitives. The hash itself is order-stable across all callers; the lockstep ordering is purely * for readability when reasoning about lookup and equality in tandem. + * + *

{@code peerTags} is taken as a {@code (array, length)} pair so the same routine works for + * the {@link Canonical} scratch buffer (where {@code length < array.length}) and the entry's + * fixed-size array. */ static long hashOf( UTF8BytesString resource, @@ -368,7 +384,8 @@ static long hashOf( short httpStatusCode, boolean synthetic, boolean traceRoot, - List peerTags) { + UTF8BytesString[] peerTags, + int peerTagsLen) { long h = 0; h = LongHashingUtils.addToHash(h, resource); h = LongHashingUtils.addToHash(h, service); @@ -379,10 +396,8 @@ static long hashOf( h = LongHashingUtils.addToHash(h, httpMethod); h = LongHashingUtils.addToHash(h, httpEndpoint); h = LongHashingUtils.addToHash(h, grpcStatusCode); - // indexed iteration -- avoids the iterator allocation a for-each over a List would do - int peerTagCount = peerTags.size(); - for (int i = 0; i < peerTagCount; i++) { - h = LongHashingUtils.addToHash(h, peerTags.get(i)); + for (int i = 0; i < peerTagsLen; i++) { + h = LongHashingUtils.addToHash(h, peerTags[i]); } h = LongHashingUtils.addToHash(h, httpStatusCode); h = LongHashingUtils.addToHash(h, synthetic); @@ -471,7 +486,7 @@ boolean isTraceRoot() { return traceRoot; } - List getPeerTags() { + UTF8BytesString[] getPeerTags() { return peerTags; } @@ -502,11 +517,15 @@ static final class Canonical { boolean traceRoot; /** - * Reusable buffer of canonicalized peer-tag UTF8 forms. Cleared and refilled in {@link - * #populate}; on miss, {@link #toEntry} copies it into an immutable list for the entry to own. - * Zero allocation on the hit path. + * Reusable buffer of canonicalized peer-tag UTF8 forms. Slots {@code [0..peerTagsCount)} are + * the live entries; the rest is dead space. The buffer doubles when it runs out of room (rare, + * since typical peer-tag schemas have very few tags). On miss, {@link #toEntry} snapshots into + * a tight {@link UTF8BytesString}{@code []} for the entry to own. Zero allocation on the hit + * path. */ - final ArrayList peerTagsBuffer = new ArrayList<>(4); + UTF8BytesString[] peerTagsBuffer = new UTF8BytesString[4]; + + int peerTagsCount; long keyHash; @@ -539,7 +558,8 @@ void populate(SpanSnapshot s) { httpStatusCode, synthetic, traceRoot, - peerTagsBuffer); + peerTagsBuffer, + peerTagsCount); } /** @@ -550,7 +570,7 @@ void populate(SpanSnapshot s) { * whenever a span carries only a subset of the configured peer tags. */ private void populatePeerTags(PeerTagSchema schema, String[] values) { - peerTagsBuffer.clear(); + peerTagsCount = 0; if (schema == null || values == null) { return; } @@ -560,7 +580,11 @@ private void populatePeerTags(PeerTagSchema schema, String[] values) { if (value == null) { continue; } - peerTagsBuffer.add(schema.register(i, value)); + UTF8BytesString utf8 = schema.register(i, value); + if (peerTagsCount == peerTagsBuffer.length) { + peerTagsBuffer = Arrays.copyOf(peerTagsBuffer, peerTagsBuffer.length * 2); + } + peerTagsBuffer[peerTagsCount++] = utf8; } } @@ -584,20 +608,21 @@ boolean matches(AggregateEntry e) { && httpMethod.equals(e.httpMethod) && httpEndpoint.equals(e.httpEndpoint) && grpcStatusCode.equals(e.grpcStatusCode) - && peerTagsEqual(peerTagsBuffer, e.peerTags) + && peerTagsEqual(peerTagsBuffer, peerTagsCount, e.peerTags) && httpStatusCode == e.httpStatusCode && synthetic == e.synthetic && traceRoot == e.traceRoot; } - /** Indexed list comparison -- avoids the iterator a {@code List.equals} would allocate. */ - private static boolean peerTagsEqual(List a, List b) { - int n = a.size(); - if (n != b.size()) { + /** + * Length-aware indexed comparison so the scratch buffer can be compared against a tight array. + */ + private static boolean peerTagsEqual(UTF8BytesString[] a, int aLen, UTF8BytesString[] b) { + if (aLen != b.length) { return false; } - for (int i = 0; i < n; i++) { - if (!a.get(i).equals(b.get(i))) { + for (int i = 0; i < aLen; i++) { + if (!a[i].equals(b[i])) { return false; } } @@ -606,18 +631,16 @@ private static boolean peerTagsEqual(List a, List snapshottedPeerTags; - int n = peerTagsBuffer.size(); - if (n == 0) { - snapshottedPeerTags = Collections.emptyList(); - } else if (n == 1) { - snapshottedPeerTags = Collections.singletonList(peerTagsBuffer.get(0)); + UTF8BytesString[] snapshottedPeerTags; + if (peerTagsCount == 0) { + snapshottedPeerTags = EMPTY_PEER_TAGS; } else { - snapshottedPeerTags = new ArrayList<>(peerTagsBuffer); + snapshottedPeerTags = new UTF8BytesString[peerTagsCount]; + System.arraycopy(peerTagsBuffer, 0, snapshottedPeerTags, 0, peerTagsCount); } return new AggregateEntry( keyHash, diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricCardinalityLimits.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricCardinalityLimits.java index f7d91343d4b..6a0b1775c6c 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricCardinalityLimits.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricCardinalityLimits.java @@ -5,30 +5,33 @@ * values collapse to a {@code blocked_by_tracer} sentinel so they merge into one aggregate row * instead of fragmenting the table. * - *

Values are sized to the typical-service workload with headroom; "typical" estimates are noted - * inline. Raise if a workload routinely hits the sentinel; lower carries proportional memory - * savings but risks suppressing legitimate distinctions. + *

Values are sized to cover realistic workload cardinality per 10s reporting window with + * headroom -- the prior DDCache-inherited limits (RESOURCE=32, OPERATION=64, ...) were chosen for + * memory conservation and were tight enough that a single REST API with a couple hundred routes + * would exhaust the budget within seconds. Memory cost with the flat handler tables is ~20 KB + * across all 9 handlers -- negligible relative to the {@code maxAggregates}-sized entry table. */ final class MetricCardinalityLimits { private MetricCardinalityLimits() {} /** - * Distinct {@code resource.name} values per cycle. Highest-cardinality field by far: DB-query - * obfuscations, HTTP route templates, custom resources. Typical service: 30-200 unique. + * Distinct {@code resource.name} values per cycle. Highest-cardinality field: HTTP route + * templates, SQL query templates, custom resources. A web app with one parameterized route per + * controller method easily hits low hundreds. */ - static final int RESOURCE = 128; + static final int RESOURCE = 512; /** * Distinct {@code service.name} values per cycle. Local service plus downstream peer-service - * names. Microservice meshes typically reference 10-50 distinct services. + * names; sized for service-mesh hubs that fan out to many downstreams. */ - static final int SERVICE = 32; + static final int SERVICE = 128; /** * Distinct {@code operation.name} values per cycle. Names like {@code http.request}, {@code - * db.query}, etc. Typical service: 10-30 across integrations. + * db.query}, etc. One per integration kind; production services often span 30-60. */ - static final int OPERATION = 64; + static final int OPERATION = 128; /** * Distinct {@code _dd.base_service} override values per cycle. Used rarely; usually empty or one @@ -37,16 +40,16 @@ private MetricCardinalityLimits() {} static final int SERVICE_SOURCE = 16; /** - * Distinct {@code span.type} values per cycle. {@code DDSpanTypes} catalog is ~30; a single - * service usually spans 5-10 integration types. + * Distinct {@code span.type} values per cycle. {@code DDSpanTypes} catalog has ~30 known values; + * a single service typically spans 5-10 integration types. */ - static final int TYPE = 16; + static final int TYPE = 32; /** - * Distinct {@code span.kind} values per cycle. OTel defines exactly 5 (server/client/producer/ - * consumer/internal); 8 still leaves 60% headroom in case a producer invents new kinds. + * Distinct {@code span.kind} values per cycle. OTel defines 5 standard kinds (server/client/ + * producer/consumer/internal); the 16 cap leaves headroom in case producers invent new kinds. */ - static final int SPAN_KIND = 8; + static final int SPAN_KIND = 16; /** * Distinct HTTP method values per cycle. Standard verbs are 7-9; WebDAV/custom adds a few more. @@ -57,7 +60,7 @@ private MetricCardinalityLimits() {} * Distinct {@code http.endpoint} values per cycle. Path templates -- same shape as {@code * RESOURCE} for HTTP-heavy services. Only used when {@code includeEndpointInMetrics} is enabled. */ - static final int HTTP_ENDPOINT = 64; + static final int HTTP_ENDPOINT = 256; /** * Distinct gRPC status code values per cycle. gRPC spec defines exactly 17 codes (0-16); 24 diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java index 03bc1a2f993..e2dc3d6f66b 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java @@ -13,7 +13,6 @@ import datadog.trace.api.git.GitInfo; import datadog.trace.api.git.GitInfoProvider; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; -import java.util.List; import java.util.function.Function; public final class SerializingMetricWriter implements MetricWriter { @@ -183,8 +182,8 @@ public void add(AggregateEntry entry) { writer.writeUTF8(entry.getSpanKind()); writer.writeUTF8(PEER_TAGS); - final List peerTags = entry.getPeerTags(); - writer.startArray(peerTags.size()); + final UTF8BytesString[] peerTags = entry.getPeerTags(); + writer.startArray(peerTags.length); for (UTF8BytesString peerTag : peerTags) { writer.writeUTF8(peerTag); diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java index ad40a2a0b67..8878cd3a211 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java @@ -1,6 +1,7 @@ package datadog.trace.common.metrics; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; +import java.util.Arrays; import java.util.List; import java.util.Objects; import javax.annotation.Nullable; @@ -16,10 +17,11 @@ * keyed on {@link AggregateEntry#keyHash}, and no production code path invokes {@link * Object#equals}. * - *

On this branch, peer tags live as a single pre-encoded {@code List} on the - * entry (canonicalization through {@link PeerTagSchema#register} already collapsed identical - * values), so equality compares the list directly. The hash side (computed in {@link - * AggregateEntry#hashOf}) folds in the encoded list, so the contract stays consistent. + *

On this branch, peer tags live as a pre-encoded {@code UTF8BytesString[]} on the entry + * (canonicalization through {@link PeerTagSchema#register} already collapsed identical values), so + * equality compares the arrays via {@link Arrays#equals(Object[], Object[])}. The hash side + * (computed in {@link AggregateEntry#hashOf}) folds in the same array, so the contract stays + * consistent. */ public final class AggregateEntryTestUtils { private AggregateEntryTestUtils() {} @@ -75,7 +77,7 @@ public static boolean equals(AggregateEntry a, AggregateEntry b) { && Objects.equals(a.getServiceSource(), b.getServiceSource()) && Objects.equals(a.getType(), b.getType()) && Objects.equals(a.getSpanKind(), b.getSpanKind()) - && a.getPeerTags().equals(b.getPeerTags()) + && Arrays.equals(a.getPeerTags(), b.getPeerTags()) && Objects.equals(a.getHttpMethod(), b.getHttpMethod()) && Objects.equals(a.getHttpEndpoint(), b.getHttpEndpoint()) && Objects.equals(a.getGrpcStatusCode(), b.getGrpcStatusCode()); diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateTableTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateTableTest.java index 8694892ea84..a739a0f0b62 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateTableTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateTableTest.java @@ -8,6 +8,7 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; import datadog.metrics.agent.AgentMeter; import datadog.metrics.api.statsd.StatsDClient; @@ -96,6 +97,35 @@ void peerTagPairsParticipateInIdentity() { assertEquals(3, table.size()); } + @Test + void cardinalityBlockedValuesCollapseIntoOneEntry() { + // SERVICE_HANDLER has a cardinality limit of 128. With 150 distinct service names, services + // 129+ canonicalize to the "blocked_by_tracer" sentinel. Because the table hashes from the + // canonical (post-handler) form, all blocked services land in the same bucket and merge into + // a single entry rather than fragmenting. + // + // Sentinel substitution is gated on AggregateEntry#LIMITS_ENABLED. With the flag off (the + // default), over-cap values get fresh UTF8BytesStrings and flow to distinct buckets, so this + // test only meaningfully runs in limits-on mode. + assumeTrue( + AggregateEntry.LIMITS_ENABLED, + "cardinality collapse only fires when the limits flag is enabled"); + + AggregateEntry.resetCardinalityHandlers(); + AggregateTable table = new AggregateTable(256); + + for (int i = 0; i < 150; i++) { + AggregateEntry entry = table.findOrInsert(snapshot("svc-" + i, "op", "client")); + assertNotNull(entry); + entry.recordOneDuration(1L); + } + + // 128 in-budget services + 1 collapsed "blocked_by_tracer" entry = 129 total. + assertEquals(129, table.size()); + + AggregateEntry.resetCardinalityHandlers(); + } + @Test void capOverrunEvictsStaleEntry() { AggregateTable table = new AggregateTable(2); diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorBootstrapTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorBootstrapTest.java index 67857ff3830..b0572fb8af8 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorBootstrapTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorBootstrapTest.java @@ -1,7 +1,7 @@ package datadog.trace.common.metrics; import static java.util.concurrent.TimeUnit.SECONDS; -import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; @@ -284,14 +284,15 @@ void reconcileSwapsSchemaWhenTagSetChanges() throws Exception { ArgumentCaptor entryCaptor = ArgumentCaptor.forClass(AggregateEntry.class); verify(writer, times(2)).add(entryCaptor.capture()); List entries = entryCaptor.getAllValues(); - assertEquals( - Collections.singletonList(UTF8BytesString.create("peer.hostname:localhost")), + assertArrayEquals( + new UTF8BytesString[] {UTF8BytesString.create("peer.hostname:localhost")}, entries.get(0).getPeerTags(), "pre-swap snapshot should encode only peer.hostname"); - assertEquals( - Arrays.asList( - UTF8BytesString.create("peer.hostname:localhost"), - UTF8BytesString.create("peer.service:billing")), + assertArrayEquals( + new UTF8BytesString[] { + UTF8BytesString.create("peer.hostname:localhost"), + UTF8BytesString.create("peer.service:billing") + }, entries.get(1).getPeerTags(), "post-swap snapshot should encode both peer.hostname and peer.service"); diff --git a/docs/client_metrics_design.md b/docs/client_metrics_design.md index bdf24b92274..3544a5f6885 100644 --- a/docs/client_metrics_design.md +++ b/docs/client_metrics_design.md @@ -8,6 +8,83 @@ does not have to sample every span to know request rates and latencies. Code lives in package `datadog.trace.common.metrics`. +## Overview + +Tracers emit thousands of spans per second. Reporting every one to the Datadog +Agent — let alone storing them — would be expensive and mostly redundant. The +question "what's my p95 latency for `GET /users/:id` on `web-frontend` over the +last 10s" doesn't need every individual span; it needs an aggregate. + +The client-stats pipeline computes those aggregates on the tracer itself and +ships rolled-up **buckets** to the agent on a fixed cadence. Each bucket is a +tuple of label values (resource, service, operation, span kind, peer tags, http +method/endpoint/status, grpc status, ...) plus a small accumulator: hit count, +error count, top-level count, duration sum, ok-latency histogram, error-latency +histogram. A bucket spans one reporting cycle (default 10 seconds); at the end +of the cycle the buckets are serialized to the agent's `/v0.6/stats` endpoint +and the in-memory accumulators are cleared. + +### Goals + +- **Bounded memory.** The aggregator's footprint must not grow without limit no + matter how many distinct label combinations the workload produces, or how + high the span throughput is. +- **No producer-thread contention.** Application threads that complete a span + shouldn't block on a lock or do meaningful work beyond cheap field + extraction. The tracer is a guest in the application's process; it must not + show up as overhead. +- **Correctness under reset.** Cardinality budgets and histograms are dropped + every reporting cycle. Mid-cycle drops and agent-downgrade clears can't + corrupt the aggregate table or fragment a single logical bucket. +- **Stable wire format.** The bucket payload matches the existing `/v0.6/stats` + schema. This is a re-implementation of an existing protocol, not a protocol + change. + +### Architecture in one paragraph + +A single **aggregator thread** owns every piece of mutable state — the bucket +table, the per-field cardinality budgets, the histograms. Application threads +build a small immutable **span snapshot** per metrics-eligible span and post it +to a bounded MPSC inbox. The aggregator drains the inbox, **canonicalizes** +each snapshot's label values through cardinality-capped UTF8 interners, hashes +the canonical form, and finds-or-inserts a bucket. A scheduled signal flushes +buckets to the agent every reporting interval; the cardinality budgets are +reset in lockstep with the flush. + +### Three rules + +1. **Producer threads never touch shared state.** They build a snapshot and + hand it off through the inbox. The aggregator does all the heavy work + (canonicalization, hashing, lookup, accumulator updates). +2. **Cardinality is capped per field, per cycle.** Each label field has its + own budget; overflow values collapse to a single `blocked_by_tracer` + sentinel so the bucket table can never grow past + `maxAggregates × (sum of per-field budgets)` distinct combinations. +3. **Aggregation happens on canonical UTF8 forms.** Two snapshots that disagree + only on representation (same content delivered once as a `String`, once as + a `UTF8BytesString`) collapse to the same bucket because hashing and + matching happen *after* canonicalization. + +### Trade-offs + +- **One snapshot allocation per metrics-eligible span.** ~100 bytes per + snapshot; cheap individually but a meaningful share of producer allocation + at high span throughput. Snapshots could be pooled or replaced with a + struct-of-arrays inbox; neither is currently worth the complexity. +- **Cap-overrun drops the new key, not LRU.** When the bucket table is at + capacity and no entry is stale enough to evict, an incoming snapshot for a + new label combination is dropped (and reported via + `onStatsAggregateDropped`). This protects the steady-state workload from a + burst of new keys that would otherwise displace established buckets. +- **One aggregator thread.** The whole consumer side is single-threaded by + design — locks, races, and visibility footguns are confined to the producer + → inbox handoff. If the producer rate is sustainedly higher than the + aggregator can drain, the inbox fills and snapshots are dropped + (`onStatsInboxFull`). +- **Fixed bucket table.** The hashtable's bucket array is sized once at + startup from `maxAggregates`. No dynamic resizing; entries beyond the cap + trigger the drop-new-key path above. + ## High-level shape ``` @@ -240,8 +317,8 @@ Two distinct cadences: ## Memory and lifetime -- `AggregateMetric` is **not thread-safe**. It is mutated only by the - aggregator thread. +- `AggregateEntry`'s per-bucket counters + histograms are **not thread-safe**; + they are mutated only by the aggregator thread. - `AggregateTable` is **not thread-safe**. All paths (producer-side `CLEAR`, schedule-driven `REPORT`, drainer-driven inserts) route through the inbox. - `Canonical` and the cardinality handlers are aggregator-thread-only. @@ -279,6 +356,81 @@ The producer reports per-trace stats via `HealthMetrics`: | Cardinality budget exhausted | Overflow values canonicalize to a `blocked_by_tracer` sentinel and merge into one bucket. Total entry count stays bounded by `maxAggregates`. | | Producer throws mid-trace | Caught by the writer's normal error path; `onClientStatTraceComputed` is not called for that trace. | +## Behavior under adversarial load + +A useful stress test (captured as `AdversarialMetricsBenchmark`): 8 producer +threads call `publish` in a tight loop with **unique** `(service, operation, +resource, peer.hostname)` per op, random durations across 30 orders of +magnitude, and random `error` / `topLevel` flags. Every cardinality dimension +saturates within milliseconds. + +### What "OOM the metrics subsystem" would look like + +A successful attack would either grow the aggregator's heap unboundedly, or +back up the producer so a synchronous structure (cache, map) grew with each +unique label combination. The current shape rules both out by construction: + +- **Inbox is a fixed-size MPSC queue.** Overflow returns `false` from + `offer` and the producer drops the snapshot via `onStatsInboxFull`. + The snapshot becomes garbage immediately; no queue growth. +- **`AggregateTable` is a fixed-size bucket array.** Insertion when the + table is full triggers an evict-stale pass (one entry with + `hitCount == 0`); if that fails the snapshot is dropped via + `onStatsAggregateDropped`. The table never resizes. +- **Cardinality handlers are flat open-addressed arrays.** Overflow values + canonicalize to the shared `blocked_by_tracer` sentinel — same hash, + same bucket, merged in. No node allocations, no rehash. +- **Histograms use `CollapsingLowestDenseStore(1024)`.** Bucket array + caps at ~8 KB per histogram. Worst case at full table cap: 2048 entries + × 2 histograms × ~8 KB ≈ 32 MB. That's the headline upper bound. +- **Empty error histograms aren't allocated until first error + recorded.** Entries that never error keep `errorLatencies = null`, + saving the wrapper allocation. + +### Measured behavior (1f × 1wi × 3i × 15s, 8 threads each side) + +| | master (`ConflatingMetricsAggregator`) | this design (`ClientStatsAggregator`) | +|---|---:|---:| +| Iteration 1 throughput | 1,506,007 ops/s | ~5,853,917 ops/s | +| Iteration 2 throughput | 1,255,258 ops/s | ~5,800,000 ops/s | +| Iteration 3 throughput | **410,097 ops/s** (-73%) | ~5,853,917 ops/s (stable) | +| GC time / 15 s wall | iter 1: 8.7 s — iter 2: 9.8 s — iter 3: **18.6 s** (multi-thread GC saturation) | ~150 ms total | +| Producer allocation | ~1,108 B/op | ~823 B/op | +| Aggregator thread state at end | "Skipped metrics reporting because the queue is full" + thread idle waiting for inbox | Continuously draining; ~13M snapshots/sec consumed | +| Inbox-full drops | (no counter on master) | ~139 M dropped over 45 s, all reported via `onStatsInboxFull` | +| Aggregate-table drops | 0 | 0 | + +### Why master degrades + +On master, the producer does **everything** synchronously on the calling +thread: `MetricKey` canonicalization, `DDCache` lookups for each label field, +`LRUCache` insertion. There is no queue between producer and consumer +— there *is* no consumer thread for the storage work, only for the +periodic report. So a 1,108 B/op allocation rate × 8 threads × ~1.5 M ops/s +generates ~13 GB/sec of garbage on the same thread that has to keep up with +incoming spans. The young gen fills, then survivor, then old gen, then full +GC. By iteration 3 the JVM is spending more than its wall-clock budget on +GC (multiple concurrent GC threads, summed > 15 s during a 15 s window) and +throughput collapses 73 %. + +### Why this design holds + +The producer/consumer split converts allocation pressure into **backpressure +at the inbox boundary**. The producer's per-op work is just "allocate one +`SpanSnapshot`, set a few `volatile` refs, `inbox.offer`, return." On +overflow, `offer` returns `false` and the snapshot is dropped on the spot — +no waiting, no allocation amplification. The aggregator thread runs at +its natural rate (~13 M snapshots/sec on the test machine), and the gap +between producer and consumer becomes the drop rate, not heap growth. +`onStatsInboxFull` makes that gap observable so operators can size +`tracerMetricsMaxPending` and `tracerMetricsMaxAggregates` for their +workload. + +Net: under adversarial input the new design absorbs what it can compute +meaningfully and drops what it can't, with both numbers exposed via health +metrics. The bounded-design properties hold to the ~32 MB worst-case +ceiling described above. + ## Why the redesign (history) The pipeline was previously `ConflatingMetricsAggregator` with: @@ -319,8 +471,20 @@ showed the producer dominating CPU time. The major shifts: ### Benchmark summary -`ClientStatsAggregatorDDSpanBenchmark` (64 client-kind DDSpans per op, single -trace, real `CoreTracer` with a no-op writer): +Four JMH benchmarks cover the producer pipeline at different angles: + +- `ClientStatsAggregatorBenchmark` — 64 SimpleSpans per op, identical labels + (consumer-side cache hit path). +- `ClientStatsAggregatorDDSpanBenchmark` — same as above but real `DDSpan` + via `CoreTracer`; exercises the production `isKind` / cached span-kind + ordinal path. +- `ClientStatsAggregatorMissPathBenchmark` — pool of 4096 single-span + traces with unique labels; exercises miss + insert + handler saturation. +- `AdversarialMetricsBenchmark` — 8 threads, unique labels per op, random + durations + error flags; pushes every bound to its limit (see + [Behavior under adversarial load](#behavior-under-adversarial-load)). + +Optimization progression on the DDSpan benchmark: | Variant | µs/op | |---|---| @@ -329,6 +493,11 @@ trace, real `CoreTracer` with a no-op writer): | with peer-tag schema hoist | 2.410 | | with cached span-kind ordinal + isSynthetic fix | 1.995 | +On the producer-bound miss-path benchmark (single-span trace, unique +labels), `ClientStatsAggregatorMissPathBenchmark` measures **0.057 µs/op + +96 B/op** vs master's **0.305 µs/op + 399 B/op** — 5.3× faster, 4.2× less +producer-thread allocation per metrics-eligible span. + The remaining producer-thread hotspots (from JFR sampling) are tag-map lookups for `peer.hostname` / other peer-tag values inside `capturePeerTagValues`. A bulk peer-tag accessor on `DDSpan` would crack that