Skip to content

Commit 77964b3

Browse files
dougqhdevflow.devflow-routing-intake
andauthored
Defer MetricKey construction to the aggregator thread (#11381)
Trim per-span work on metrics aggregator publish path ConflatingMetricsAggregator.publish does a handful of redundant operations on every span. None individually is large; together they show as ~2.5% on the existing JMH benchmark once the benchmark actually exercises span.kind. - dedup span.isTopLevel(): publish() reads it into a local, then shouldComputeMetric read it again. Pass the cached value in. - resolve spanKind to String once: master called toString() twice per span (once inside spanKindEligible, once at the getPeerTags call site) and used HashSet contains on a CharSequence (which routes through equals on String). Normalize to String up front and reuse. - lazy-allocate the peer-tag list: getPeerTags() always allocated an ArrayList sized to features.peerTags() even when the span had none of those tags set. Defer allocation until the first match; return Collections.emptyList() when none hit. MetricKey already treats null/empty peerTags as emptyList, so no behavior change. Drop the spanKindEligible helper — the HashSet.contains call inlines fine in shouldComputeMetric. Update the JMH benchmark to set span.kind=client on every span. Without it the filter path short-circuits before the peer-tag and toString work, so the wins above aren't measurable. With it: baseline 6.755 us/op (CI [6.560, 6.950], stdev 0.129) optimized 6.585 us/op (CI [6.536, 6.634], stdev 0.033) 2 forks x 5 iterations x 15s. ~2.5% mean improvement and much tighter variance fork-to-fork. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Add SpanKindFilter and CoreSpan.isKind for bitmask-based kind checks Introduce SpanKindFilter -- a tiny builder-built immutable filter whose state is an int bitmask indexed by the span.kind ordinals already cached on DDSpanContext. Each include* on the builder sets one bit (1 << ordinal); the runtime check is a single AND against (1 << span's ordinal). CoreSpan.isKind(SpanKindFilter) is the new entry point. DDSpan overrides it to do the bit-test directly against the cached ordinal -- no virtual call, no tag-map lookup. The two existing test-only CoreSpan impls (SimpleSpan and TraceGenerator.PojoSpan, the latter in two source sets) implement isKind by reading the span.kind tag and delegating to SpanKindFilter.matches(String), which converts via DDSpanContext.spanKindOrdinalOf and does the same AND. Refactor: DDSpanContext.setSpanKindOrdinal(String) now delegates to a new package-private static spanKindOrdinalOf(String) so the same string-to-ordinal mapping serves both the tag interceptor path and SpanKindFilter.matches. This is groundwork -- nothing in the codebase calls isKind yet. The next commit will replace the HashSet-based eligibility checks in ConflatingMetricsAggregator with SpanKindFilter instances. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Use SpanKindFilter in ConflatingMetricsAggregator Replace the two ELIGIBLE_SPAN_KINDS_FOR_* HashSet<String> constants and the SPAN_KIND_INTERNAL.equals check with three SpanKindFilter instances: METRICS_ELIGIBLE_KINDS, PEER_AGGREGATION_KINDS, INTERNAL_KIND. Eligibility checks now go through span.isKind(filter), which on DDSpan is a volatile byte read against the already-cached span.kind ordinal plus a single bit-test. Also defer the span.kind tag read: previously read at the top of the publish loop and threaded through both shouldComputeMetric and the inner publish. isKind no longer needs the string, so the read can move down into the inner publish where it's still needed for the SPAN_KINDS cache key / MetricKey. Supporting changes: - DDSpanContext.spanKindOrdinalOf(String) is now public so non-DDSpan CoreSpan impls can compute the ordinal at tag-write time. - SpanKindFilter gains a public matches(byte) fast-path overload that callers with a pre-computed ordinal use directly. - SimpleSpan caches the ordinal in setTag(SPAN_KIND, ...), mirroring what TagInterceptor does for DDSpanContext, and its isKind now hits the byte fast path. Without this, the JMH benchmark (which uses SimpleSpan) would re-derive the ordinal on every isKind call and overstate the cost. Benchmark on the bench updated last commit (kind=client on every span, 4 forks x 5 iter x 15s): prior commit 6.585 ± 0.049 us/op this commit 6.903 ± 0.096 us/op The slight regression is a SimpleSpan-via-groovy-dispatch artifact -- the interface call to isKind through CoreSpan, then through SimpleSpan, then through SpanKindFilter.matches, doesn't fold as aggressively as a HashSet contains on a static field. In production DDSpan.isKind inlines to a context field read + ordinal byte read + bit-test, so the production path is faster than the prior HashSet approach. A DDSpan-based benchmark would show this; the existing SimpleSpan-based one doesn't. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Add DDSpan-based variant of ConflatingMetricsAggregator JMH benchmark The existing ConflatingMetricsAggregatorBenchmark uses SimpleSpan, a groovy mock. That's enough for measuring queue/CHM/MetricKey work, but it conceals the production cost of CoreSpan.isKind: SimpleSpan's isKind goes through groovy interface dispatch into SpanKindFilter.matches, while DDSpan.isKind inlines to a context byte-read + bit-test. This new benchmark uses real DDSpan instances created through a CoreTracer (with a NoopWriter so finishing doesn't reach the agent). Same shape as the SimpleSpan bench (64-span trace, span.kind=client, peer.hostname set). Numbers (2 forks x 5 iter x 15s): master: 6.428 +- 0.189 us/op (HashSet eligibility checks) this branch: 6.343 +- 0.115 us/op (SpanKindFilter bitmask) About 1.3% faster on the production path. The SimpleSpan benchmark in the same conditions shows a ~2.2% slowdown -- the mock's dispatch shape gives a misleading signal. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Tighten SpanKindFilter encapsulation Make SpanKindFilter.kindMask and its constructor private now that DDSpan.isKind no longer needs direct field access -- it delegates to SpanKindFilter.matches(byte). The Builder.build() in the same outer class still constructs instances via the private constructor. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Defer MetricKey construction and cache lookups to the aggregator thread Replace the producer-side conflation pipeline with a thin per-span SpanSnapshot posted to the existing aggregator thread. The aggregator now builds the MetricKey, does the SERVICE_NAMES / SPAN_KINDS / PEER_TAGS_CACHE lookups, and updates the AggregateMetric directly -- all off the producer's hot path. What the producer does now, per span: - filter (shouldComputeMetric, resource-ignored, longRunning) - collect tag values into a SpanSnapshot (1 allocation per span) - inbox.offer(snapshot) + return error flag for forceKeep What moved off the producer: - MetricKey construction and its hash computation - SERVICE_NAMES.computeIfAbsent (UTF8 encoding of service name) - SPAN_KINDS.computeIfAbsent (UTF8 encoding of span.kind) - PEER_TAGS_CACHE lookups (peer-tag name+value UTF8 encoding) - pending/keys ConcurrentHashMap operations - Batch pooling, batch atomic ops, batch contributeTo Removed entirely: - Batch.java -- the conflation primitive is no longer needed; the aggregator's existing LRUCache<MetricKey, AggregateMetric> IS the conflation point now. - pending ConcurrentHashMap<MetricKey, Batch> - keys ConcurrentHashMap<MetricKey, MetricKey> (canonical dedup) - batchPool MessagePassingQueue<Batch> - The CommonKeyCleaner role of tracking keys.keySet() on LRU eviction -- AggregateExpiry now just reports drops to healthMetrics. Added: - SpanSnapshot: immutable value carrying the raw MetricKey inputs + a tagAndDuration long (duration | ERROR_TAG | TOP_LEVEL_TAG). - AggregateMetric.recordOneDuration(long tagAndDuration) -- the single-hit equivalent of the existing recordDurations(int, AtomicLongArray). - Peer-tag values flow through the snapshot as a flattened String[] of [name0, value0, name1, value1, ...]; the aggregator encodes them through PEER_TAGS_CACHE on its own thread. Benchmark results (2 forks x 5 iter x 15s): ConflatingMetricsAggregatorDDSpanBenchmark prior commit 6.343 +- 0.115 us/op this commit 2.506 +- 0.044 us/op (~60% faster) ConflatingMetricsAggregatorBenchmark (SimpleSpan) prior commit 6.585 +- 0.049 us/op this commit 3.116 +- 0.032 us/op (~53% faster) Caveat on the benchmark: without conflation, the producer pushes 1 inbox item per span instead of ~1 per 64. At the benchmark's synthetic rate the consumer can't keep up and inbox.offer silently drops. The numbers measure producer publish() latency only; consumer throughput at realistic span rates is a follow-up to validate. Tuning maxPending matters more in this design. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Report aggregator inbox-full drops via health metrics With the per-span SpanSnapshot inbox path, the producer can lose snapshots when the bounded MPSC queue is full -- silently, since inbox.offer() returns a boolean we previously ignored. The conflating-Batch design used to absorb ~64x more producer pressure per inbox slot, so this is a new failure mode worth surfacing. Wire it through the existing HealthMetrics path: - HealthMetrics.onStatsInboxFull() (no-op default). - TracerHealthMetrics gets a statsInboxFull LongAdder and a new reason tag reason:inbox_full reported under the same stats.dropped_aggregates metric used for LRU evictions. Two LongAdders, two tagged time series. - ConflatingMetricsAggregator.publish increments the counter when inbox.offer(snapshot) returns false. This doesn't fix the drop -- tuning maxPending and/or building producer-side batching are the actual fixes. But it makes the failure visible in the same place ops already watches. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge branch 'master' into dougqh/conflating-metrics-producer-wins Merge branch 'dougqh/conflating-metrics-producer-wins' into dougqh/conflating-metrics-background-work Resize previousCounts for inbox-full health metric The new reason:inbox_full reportIfChanged call advances countIndex to 51, but previousCounts was still sized for 51 counters (max index 50), so the metric never emitted and the resize warning fired every flush. Bump the array to 52 and add a regression test that exercises the flush path. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Skip SpanSnapshot allocation when the inbox is already at capacity publish() previously did all of the tag extraction (peer-tag pairs, HTTP method/endpoint, span kind, gRPC status) and the SpanSnapshot allocation before calling inbox.offer; on a full inbox the offer failed and everything became garbage. Early-out with an approximate size() vs capacity() check up front. The jctools MPSC queue's size() is best-effort but that's fine: under- estimation falls through to the existing offer-as-source-of-truth path, over-estimation drops a snapshot that would have fit (and onStatsInboxFull was about to fire on the next span anyway). error is computed first so the force-keep return is correct whether or not the snapshot is built. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge remote-tracking branch 'origin/master' into dougqh/conflating-metrics-background-work Introduce slim PeerTagSchema; capture peer-tag values not pairs Addresses sarahchen6's review comment on ConflatingMetricsAggregator extractPeerTagPairs: replaces the worst-case-allocation + trim-and-copy flat-pairs layout with a parallel-array carrier. - New PeerTagSchema: minimal carrier of String[] names. Two flavors -- a static INTERNAL singleton (one entry: base.service) for internal-kind spans, and per-discovery built schemas for client/producer/consumer spans. Deliberately no cardinality limiters or per-cycle state; that layers on top in a later PR. - ConflatingMetricsAggregator: caches the peer-aggregation schema keyed on reference equality of features.peerTags() -- a single volatile read + a long compare on the steady-state producer hot path, no allocation. The producer now captures only a String[] of values parallel to the schema's names; the schema reference is carried on SpanSnapshot. The prior "build worst-case pairs then trim" code is gone. - SpanSnapshot: replaces String[] peerTagPairs with PeerTagSchema + String[] peerTagValues. Producer drops the schema reference if no values fired so the consumer short-circuits on null. - Aggregator.materializePeerTags: now reads name/value pairs at the same index from (schema.names, snapshot.peerTagValues). Counts hits once for exact-size allocation; preserves the singletonList fast path for the common one-entry case (e.g. internal-kind base.service). Producer-side cost goes from "allocate String[2n] + walk + maybe trim" to "single volatile read + walk + lazy String[n] only on first hit". Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Address PR #11381 review (round 2) - Aggregator.materializePeerTags: fold the firstHit-discovery nested if into a single guarded post-increment (amarziali, #3279243138). One body line: `if (values[i] != null && hitCount++ == 0) firstHit = i;`. - Drop redundant isKind(SpanKindFilter) overrides in both TraceGenerator.groovy files (amarziali, #3279264553 / #3279382648). CoreSpan.java:84 already supplies a default implementation that reads the same span.kind tag. - Bump TRACER_METRICS_MAX_PENDING default from 2048 -> 131072 to address the capacity regression amarziali flagged (#3279378375). Without producer-side conflation, the inbox now holds 1 SpanSnapshot per metrics-eligible span instead of 1 conflated Batch per ~64 spans; restoring effective capacity parity (~2048 * ~64 = 131072) prevents a ~64x rise in inbox-full drops at the same span rate. ~100 B per SpanSnapshot puts the worst-case heap floor at ~13 MB -- bounded. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Cover inbox-full fast-path in ConflatingMetricsAggregator.publish Addresses PR #11381 review (amarziali, #3279325340 -- "Are the existing tests covering this case?"). New ConflatingMetricsAggregatorInboxFullTest constructs the aggregator with a small inbox (queueSize=8), deliberately does NOT call start() so the consumer thread never drains, then publishes enough spans to overflow the inbox. Verifies that healthMetrics.onStatsInboxFull() is called at least once -- the fast-path's `inbox.size() >= inbox.capacity()` short-circuit triggers when the producer-side queue is at capacity. Test is Java + JUnit 5 + Mockito per the project convention for new tests; uses a CoreSpan Mockito mock rather than the SimpleSpan Groovy fixture so we don't depend on Groovy-then-Java compile order from the test source set. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Reconcile PeerTagSchema once per reporting cycle on the aggregator thread Addresses amarziali's review comment #3279340181 ("It would be more efficient to trigger from the other side"). The producer-side reference compare on every publish goes away; the aggregator thread reconciles the cached schema against feature discovery once per reporting cycle. - DDAgentFeaturesDiscovery: expose getLastTimeDiscovered() so callers can detect a discovery refresh without copying the peerTags Set. - PeerTagSchema: add `long lastTimeDiscovered` (plain, aggregator-only) and `hasSameTagsAs(Set)`. of(Set, long) takes the timestamp; INTERNAL uses a -1L sentinel since it's never reconciled. - ConflatingMetricsAggregator: * Drop the cachedPeerTagsSource volatile and the per-publish reference compare. * Producer fast path is now `cachedPeerTagSchema` volatile read + null-check; first publish takes the one-time synchronized bootstrap. * Add reconcilePeerTagSchema() that runs once per cycle on the aggregator thread: fast-path timestamp compare, slow-path set compare, bump-in-place when the set is unchanged. - Aggregator: new `Runnable onReportCycle` constructor parameter, run at the start of report() (before the flush, so any test awaiting writer.finishBucket() observes the schema in its post-reconcile state and so the next publish sees the new schema without a handoff). - Update "should create bucket for each set of peer tags" to drive two reporting cycles separated by a report() that triggers reconcile. The old test relied on per-publish reference detection, which the new design intentionally doesn't preserve -- the schema is now stable within a cycle. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Add bootstrap + reconcile coverage for PeerTagSchema Addresses round-3 review nice-to-haves on PR #11381. - PeerTagSchemaTest: unit coverage for hasSameTagsAs() (the predicate that drives the reconcile fast/slow path split), the of(Set, long) factory, and the INTERNAL singleton. The hasSameTagsAs cases include same-content-different-Set-reference (the case the reconcile fast path relies on after a discovery refresh) and content-mismatch in either direction. - ConflatingMetricsAggregatorBootstrapTest: integration coverage for the producer-side bootstrap + aggregator-thread reconcile flow. * bootstrapHappensOnceOnFirstPublish -- three publishes against an un-started aggregator (no consumer thread, no reconciles); verifies features.peerTags() and features.getLastTimeDiscovered() are each called exactly once. * reconcileSkipsDeepCompareWhenTimestampMatches -- two cycles with constant features.getLastTimeDiscovered(); each post-report reconcile short-circuits on the timestamp fast path, so peerTags() is called only by bootstrap (1 total). * reconcileSurvivesTimestampBumpWhenTagsUnchanged -- timestamps bump every reconcile, forcing the slow set-compare path; the tag set stays identical, so the schema is preserved and continues to flush buckets correctly across cycles. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Use writer.finishBucket() count in bootstrap test for cascade compatibility The verify(writer).add(MetricKey, AggregateMetric) signature is unique to #11381; downstream branches use AggregateEntry. Switching to verify(writer, times(2)).finishBucket() keeps the same behavioral guarantee (both cycles flushed) across the stack. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge branch 'master' into dougqh/conflating-metrics-background-work Preserve TRACER_METRICS_MAX_PENDING semantic + drop stale imports TRACER_METRICS_MAX_PENDING previously counted conflating Batch slots (~64 spans each). The inbox now holds 1 SpanSnapshot per slot, so multiply the configured value by LEGACY_BATCH_SIZE (64) to keep pre-existing customer overrides delivering the same effective span-throughput capacity. Default stays at 2048 logical -> 131072 snapshot slots, identical to the prior 2048 batches * 64 spans. Also drops two unused datadog.trace.core.SpanKindFilter imports left behind in TraceGenerator.groovy after the isKind() override was removed in favor of the CoreSpan default implementation. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Add AdversarialMetricsBenchmark for capacity-bound stress testing Ports the adversarial JMH benchmark from #11402 down to this branch so we can compare #11381 vs master on a high-cardinality, high-throughput workload. Adapted to use ConflatingMetricsAggregator (pre-rename) and the FixedAgentFeaturesDiscovery / NullSink helpers already in ConflatingMetricsAggregatorBenchmark. 8 producer threads hammer publish() with unique (service, operation, resource, peer.hostname) per op so the aggregate cache fills+evicts continuously and the inbox saturates. tearDown prints the drop counters (inboxFull vs aggregateDropped) so the test verifies the subsystem stayed bounded under attack. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Trim AdversarialMetricsBenchmark counters and clarify printout Drop traceComputedCalls / totalSpansCounted: under 8-way contention the volatile-long ++/+= pattern was losing ~20% of updates (296M counted vs 245M reported), and the numbers duplicate signal JMH's ops/s already provides. Switch inboxFull / aggregateDropped to LongAdder so the printed drop shape (the order-of-magnitude story the bench is built to tell) is accurate under contention. Replace the stale "both forks combined for this run" string with text that matches the actual @fork(value=1) config and notes that counters accumulate across warmup + measurement. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Close PeerTagSchema reconcile race + cover the swap branch buildPeerTagSchema previously read features.peerTags() before features.getLastTimeDiscovered(). DDAgentFeaturesDiscovery exposes those as two separate accessors against its volatile State -- a state-swap interleaving could leave the cached schema tagged with a NEWER timestamp than its names, after which the next reconcile short-circuits on the timestamp compare and misses the tag-set update until the next discovery refresh (~minute later). Swap the read order so timestamp is captured first. With this ordering, an interleaving leaves the schema OLDER than its names instead -- the next reconcile sees a timestamp mismatch, runs the deep compare, and self-heals on the very next cycle. Also adds reconcileSwapsSchemaWhenTagSetChanges, which closes the test gap on the slow-path swap branch (cachedPeerTagSchema = PeerTagSchema.of(...)). End-to-end check via the writer's captured MetricKeys: pre-swap snapshot carries only peer.hostname, post-swap snapshot carries both peer.hostname and peer.service. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Clarify materializePeerTags hit-counting loop Splits the `if (values[i] != null && hitCount++ == 0)` conjunction into nested ifs. Same semantics, no codegen impact after JIT -- just visibly says what the loop is doing rather than relying on post-increment-inside-conjunction. Closes amarziali's review thread on this block. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Drop unused Tags imports flagged by codenarc Leftover from removing the isKind() override in TraceGenerator earlier in this session -- I dropped the SpanKindFilter import but missed datadog.trace.bootstrap.instrumentation.api.Tags, which is no longer referenced in either file. Resolves codenarcTest and codenarcTraceAgentTest UnusedImport violations. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Update dd-trace-core/src/main/java/datadog/trace/common/metrics/PeerTagSchema.java Co-authored-by: Sarah Chen <sarah.chen@datadoghq.com> Address sarahchen6's review pass PeerTagSchema.java: drop the duplicate Javadoc line that the GitHub UI suggestion accept inadvertently added (it added rather than replaced), collapsing back to the single intended line per sarahchen6's suggestion. Original line said "no cardinality limiters or per-cycle state" which was misleading since lastTimeDiscovered IS per-cycle state; suggestion rightly drops that clause. Config.java: wrap the TRACER_METRICS_MAX_PENDING * LEGACY_BATCH_SIZE multiplication in Math.multiplyExact to fail fast on absurd customer overrides (>= ~33M) rather than silently wrap to a negative int and explode the MPSC queue allocation with a confusing downstream error. Per sarahchen6's suggestion citing the codex bot. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Clamp TRACER_METRICS_MAX_PENDING instead of throwing on overflow The previous Math.multiplyExact approach would fail the agent startup with ArithmeticException on absurd customer overrides (>= ~33M for the configured value). Clamping is gentler -- the agent starts successfully and just runs with a capped inbox. Long-promote the multiplication to a long so the product can't wrap, then clamp to MAX_SAFE_ARRAY_SIZE (Integer.MAX_VALUE - 8, the JDK's own SOFT_MAX_ARRAY_LENGTH convention for array allocations). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge branch 'master' into dougqh/conflating-metrics-background-work Suppress forbiddenApis for tearDown's System.err diagnostics AdversarialMetricsBenchmark.tearDown prints drop counters via System.err so a benchmark run shows how saturated each capacity bound was (inbox-full drops, aggregate-cache drops). forbiddenApisJmh disallows System.err by default to prevent excess logging in production code -- not a concern for a JMH benchmark, where stderr is the conventional channel for diagnostic output and matches the existing pattern in ExtractorBenchmark / InjectorBenchmark. Annotates tearDown with @SuppressForbidden (method-scoped, not class- scoped) so the suppression is narrowly targeted to the three println calls and any future hot-path code that lands in the benchmark stays gated by the check. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge branch 'master' into dougqh/conflating-metrics-background-work Use DDAgentFeaturesDiscovery.state() hash for PeerTagSchema reconcile Addresses amarziali's review on getLastTimeDiscovered(): the existing state() accessor returns a SHA-256 of the discovery response, which is a more precise change key than the timestamp. Timestamp advances on every successful refresh regardless of content; the hash only advances when something actually changed -- so reconcile fast-path now fires only on real change, not every cycle. - PeerTagSchema: long lastTimeDiscovered -> String state. Factory signature of(Set, long) -> of(Set, String). INTERNAL carries null (it is never reconciled). - ConflatingMetricsAggregator: read features.state() first then peerTags() (same defensive ordering rationale -- if a discovery refresh interleaves, leave the schema with stale state rather than stale tags so the next reconcile re-runs the deep compare). Objects.equals for null-tolerant comparison (state can be null before discovery has produced a response). - DDAgentFeaturesDiscovery: drop the public getLastTimeDiscovered() accessor added on this branch -- the field stays private for the existing throttling logic in discoverIfOutdated(). - Tests updated to mock state() instead of getLastTimeDiscovered(). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Convert TRACER_METRICS_MAX_PENDING rationale to /* */ block comment Addresses amarziali's readability nit (#3289149416) -- multi-line prose reads better as a single block comment than as a stack of // lines. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge remote-tracking branch 'origin/dougqh/conflating-metrics-background-work' into dougqh/conflating-metrics-background-work Add cardinality-isolation companions to AdversarialMetricsBenchmark Two new JMH benches that hold every dimension constant except one, to attribute throughput deltas to a specific axis: - HighCardinalityResourceMetricsBenchmark: ~1M distinct resource values; service/operation/peer.hostname pinned. Exercises the aggregate-cache LRU on the resource axis specifically. - HighCardinalityPeerMetricsBenchmark: ~32K distinct peer.hostname values; service/operation/resource pinned. Isolates the peer-tag encoding hot path (PEER_TAGS_CACHE lookups, UTF8 encoding, parallel-array capture in SpanSnapshot). Same shape as AdversarialMetricsBenchmark (8 threads, 2x15s warmup + 5x15s measurement, 1 fork) and reuse its CountingHealthMetrics so the inbox-full vs aggregate-dropped counters print on teardown for an apples-to-apples comparison. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Merge branch 'master' into dougqh/conflating-metrics-background-work Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>
1 parent 4205397 commit 77964b3

19 files changed

Lines changed: 1458 additions & 340 deletions
Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
package datadog.trace.common.metrics;
2+
3+
import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND;
4+
import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT;
5+
import static java.util.concurrent.TimeUnit.SECONDS;
6+
7+
import datadog.trace.api.WellKnownTags;
8+
import datadog.trace.core.CoreSpan;
9+
import datadog.trace.core.monitor.HealthMetrics;
10+
import de.thetaphi.forbiddenapis.SuppressForbidden;
11+
import java.util.Collections;
12+
import java.util.List;
13+
import java.util.concurrent.ThreadLocalRandom;
14+
import java.util.concurrent.atomic.LongAdder;
15+
import org.openjdk.jmh.annotations.Benchmark;
16+
import org.openjdk.jmh.annotations.BenchmarkMode;
17+
import org.openjdk.jmh.annotations.Fork;
18+
import org.openjdk.jmh.annotations.Measurement;
19+
import org.openjdk.jmh.annotations.Mode;
20+
import org.openjdk.jmh.annotations.OutputTimeUnit;
21+
import org.openjdk.jmh.annotations.Scope;
22+
import org.openjdk.jmh.annotations.Setup;
23+
import org.openjdk.jmh.annotations.State;
24+
import org.openjdk.jmh.annotations.TearDown;
25+
import org.openjdk.jmh.annotations.Threads;
26+
import org.openjdk.jmh.annotations.Warmup;
27+
import org.openjdk.jmh.infra.Blackhole;
28+
29+
/**
30+
* Adversarial JMH benchmark designed to stress the metrics subsystem's capacity bounds.
31+
*
32+
* <p>The metrics aggregator is bounded at every layer:
33+
*
34+
* <ul>
35+
* <li>The aggregate cache caps total entries at {@code tracerMetricsMaxAggregates} (default
36+
* 2048). Beyond that LRU eviction kicks in.
37+
* <li>The producer/consumer inbox is a fixed-size MPSC queue ({@code tracerMetricsMaxPending});
38+
* when full, producer {@code offer} returns false and the snapshot is dropped via {@link
39+
* HealthMetrics#onStatsInboxFull()}.
40+
* <li>Histograms use a bounded dense store -- per-histogram memory is fixed.
41+
* </ul>
42+
*
43+
* <p>The benchmark hammers all of these simultaneously with 8 producer threads, unique labels per
44+
* op (so the aggregate cache fills+evicts repeatedly), random durations across a wide range (so
45+
* histograms accept many distinct bins), and random {@code error}/{@code topLevel} flags (so both
46+
* histograms are exercised). After the run, drop counters are printed so you can see how the
47+
* subsystem absorbed the burst.
48+
*
49+
* <p>What "OOM the metrics subsystem" would look like if the bounds break: producer-thread
50+
* allocation would grow unbounded (snapshots faster than the inbox can drain produces dropped
51+
* snapshots, not heap growth); aggregator-thread heap would grow if entries weren't capped or
52+
* histograms grew past their dense-store limit.
53+
*/
54+
@State(Scope.Benchmark)
55+
@Warmup(iterations = 2, time = 15, timeUnit = SECONDS)
56+
@Measurement(iterations = 5, time = 15, timeUnit = SECONDS)
57+
@BenchmarkMode(Mode.Throughput)
58+
@OutputTimeUnit(SECONDS)
59+
@Threads(8)
60+
@Fork(value = 1)
61+
public class AdversarialMetricsBenchmark {
62+
63+
private ConflatingMetricsAggregator aggregator;
64+
private CountingHealthMetrics health;
65+
66+
@State(Scope.Thread)
67+
public static class ThreadState {
68+
int cursor;
69+
}
70+
71+
@Setup
72+
public void setup() {
73+
this.health = new CountingHealthMetrics();
74+
this.aggregator =
75+
new ConflatingMetricsAggregator(
76+
new WellKnownTags("", "", "", "", "", ""),
77+
Collections.emptySet(),
78+
new ConflatingMetricsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
79+
Collections.singleton("peer.hostname"), Collections.emptySet()),
80+
this.health,
81+
new ConflatingMetricsAggregatorBenchmark.NullSink(),
82+
2048,
83+
2048,
84+
false);
85+
this.aggregator.start();
86+
}
87+
88+
@TearDown
89+
@SuppressForbidden
90+
public void tearDown() {
91+
aggregator.close();
92+
// Counters accumulate across the trial (warmup + measurement iterations), since the
93+
// CountingHealthMetrics instance is created once in @Setup and never reset.
94+
System.err.println(
95+
"[ADVERSARIAL] drops over the trial (8 threads, warmup + measurement combined):");
96+
System.err.println(
97+
" onStatsInboxFull = "
98+
+ health.inboxFull.sum()
99+
+ " (snapshots dropped because the MPSC inbox was full)");
100+
System.err.println(
101+
" onStatsAggregateDropped = "
102+
+ health.aggregateDropped.sum()
103+
+ " (snapshots dropped because the aggregate cache was full with no stale entry)");
104+
}
105+
106+
@Benchmark
107+
public void publish(ThreadState ts, Blackhole blackhole) {
108+
int idx = ts.cursor++;
109+
ThreadLocalRandom rng = ThreadLocalRandom.current();
110+
111+
// Mix indices so labels don't fall into linear order. Distinct labels exceed every reasonable
112+
// working-set bound, so the aggregate cache evicts continuously and most ops force a fresh
113+
// MetricKey construction on the consumer thread.
114+
int scrambled = idx * 0x9E3779B1; // golden ratio multiplier
115+
String service = "svc-" + (scrambled & 0xFFFF);
116+
String operation = "op-" + ((scrambled >>> 8) & 0x3FFFF);
117+
String resource = "res-" + ((scrambled ^ 0x5A5A5A) & 0xFFFFF);
118+
String hostname = "host-" + ((scrambled >>> 12) & 0x7FFF);
119+
boolean error = (idx & 7) == 0;
120+
boolean topLevel = (idx & 3) == 0;
121+
// Wide duration spread forces histogram bins to populate broadly.
122+
long durationNanos = 1L + (rng.nextLong() & 0x3FFFFFFFL); // 1 ns .. ~1.07 s
123+
124+
SimpleSpan span =
125+
new SimpleSpan(
126+
service, operation, resource, "web", true, topLevel, error, 0, durationNanos, 200);
127+
span.setTag(SPAN_KIND, SPAN_KIND_CLIENT);
128+
span.setTag("peer.hostname", hostname);
129+
130+
List<CoreSpan<?>> trace = Collections.singletonList(span);
131+
blackhole.consume(aggregator.publish(trace));
132+
}
133+
134+
/**
135+
* Counts what gets dropped. Uses {@link LongAdder} so the printed totals hold up under 8-way
136+
* contention -- {@code volatile long ++} loses ~20% of updates here, which would mask the
137+
* order-of-magnitude shape the bench is trying to surface (inbox-full vs aggregate-dropped).
138+
*/
139+
static final class CountingHealthMetrics extends HealthMetrics {
140+
final LongAdder inboxFull = new LongAdder();
141+
final LongAdder aggregateDropped = new LongAdder();
142+
143+
@Override
144+
public void onStatsInboxFull() {
145+
inboxFull.increment();
146+
}
147+
148+
@Override
149+
public void onStatsAggregateDropped() {
150+
aggregateDropped.increment();
151+
}
152+
}
153+
}
Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
package datadog.trace.common.metrics;
2+
3+
import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND;
4+
import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT;
5+
import static java.util.concurrent.TimeUnit.SECONDS;
6+
7+
import datadog.trace.api.WellKnownTags;
8+
import datadog.trace.common.metrics.AdversarialMetricsBenchmark.CountingHealthMetrics;
9+
import datadog.trace.core.CoreSpan;
10+
import de.thetaphi.forbiddenapis.SuppressForbidden;
11+
import java.util.Collections;
12+
import java.util.List;
13+
import java.util.concurrent.ThreadLocalRandom;
14+
import org.openjdk.jmh.annotations.Benchmark;
15+
import org.openjdk.jmh.annotations.BenchmarkMode;
16+
import org.openjdk.jmh.annotations.Fork;
17+
import org.openjdk.jmh.annotations.Measurement;
18+
import org.openjdk.jmh.annotations.Mode;
19+
import org.openjdk.jmh.annotations.OutputTimeUnit;
20+
import org.openjdk.jmh.annotations.Scope;
21+
import org.openjdk.jmh.annotations.Setup;
22+
import org.openjdk.jmh.annotations.State;
23+
import org.openjdk.jmh.annotations.TearDown;
24+
import org.openjdk.jmh.annotations.Threads;
25+
import org.openjdk.jmh.annotations.Warmup;
26+
import org.openjdk.jmh.infra.Blackhole;
27+
28+
/**
29+
* Cardinality-isolation companion to {@link AdversarialMetricsBenchmark}: only the {@code
30+
* peer.hostname} tag value varies; {@code service}, {@code operation}, and {@code resource} are
31+
* pinned to single values. Pairing this with the adversarial bench (all four dimensions
32+
* high-cardinality) and {@link HighCardinalityResourceMetricsBenchmark} (only resource
33+
* high-cardinality) lets you attribute any throughput delta to a specific axis.
34+
*
35+
* <p>This isolates the peer-tag-encoding hot path: {@code PEER_TAGS_CACHE} lookups, the per-tag
36+
* UTF8 encoding of {@code "name:value"}, and the parallel-array capture inside the producer's
37+
* {@code SpanSnapshot} build. With {@code 0x7FFF} (~32K) distinct hostnames the cache thrashes
38+
* heavily and exceeds the default {@code tracerMetricsMaxAggregates=2048} so the LRU evicts
39+
* continuously.
40+
*
41+
* <p>Random {@code error}/{@code topLevel}/duration to keep histogram load comparable; only the
42+
* cardinality profile changes.
43+
*/
44+
@State(Scope.Benchmark)
45+
@Warmup(iterations = 2, time = 15, timeUnit = SECONDS)
46+
@Measurement(iterations = 5, time = 15, timeUnit = SECONDS)
47+
@BenchmarkMode(Mode.Throughput)
48+
@OutputTimeUnit(SECONDS)
49+
@Threads(8)
50+
@Fork(value = 1)
51+
public class HighCardinalityPeerMetricsBenchmark {
52+
53+
private ConflatingMetricsAggregator aggregator;
54+
private CountingHealthMetrics health;
55+
56+
@State(Scope.Thread)
57+
public static class ThreadState {
58+
int cursor;
59+
}
60+
61+
@Setup
62+
public void setup() {
63+
this.health = new CountingHealthMetrics();
64+
this.aggregator =
65+
new ConflatingMetricsAggregator(
66+
new WellKnownTags("", "", "", "", "", ""),
67+
Collections.emptySet(),
68+
new ConflatingMetricsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
69+
Collections.singleton("peer.hostname"), Collections.emptySet()),
70+
this.health,
71+
new ConflatingMetricsAggregatorBenchmark.NullSink(),
72+
2048,
73+
2048,
74+
false);
75+
this.aggregator.start();
76+
}
77+
78+
@TearDown
79+
@SuppressForbidden
80+
public void tearDown() {
81+
aggregator.close();
82+
System.err.println(
83+
"[HIGH_CARD_PEER] drops over the trial (8 threads, warmup + measurement combined):");
84+
System.err.println(" onStatsInboxFull = " + health.inboxFull.sum());
85+
System.err.println(" onStatsAggregateDropped = " + health.aggregateDropped.sum());
86+
}
87+
88+
@Benchmark
89+
public void publish(ThreadState ts, Blackhole blackhole) {
90+
int idx = ts.cursor++;
91+
ThreadLocalRandom rng = ThreadLocalRandom.current();
92+
93+
int scrambled = idx * 0x9E3779B1;
94+
String hostname = "host-" + ((scrambled >>> 12) & 0x7FFF);
95+
boolean error = (idx & 7) == 0;
96+
boolean topLevel = (idx & 3) == 0;
97+
long durationNanos = 1L + (rng.nextLong() & 0x3FFFFFFFL);
98+
99+
SimpleSpan span =
100+
new SimpleSpan("svc", "op", "res", "web", true, topLevel, error, 0, durationNanos, 200);
101+
span.setTag(SPAN_KIND, SPAN_KIND_CLIENT);
102+
span.setTag("peer.hostname", hostname);
103+
104+
List<CoreSpan<?>> trace = Collections.singletonList(span);
105+
blackhole.consume(aggregator.publish(trace));
106+
}
107+
}
Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
package datadog.trace.common.metrics;
2+
3+
import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND;
4+
import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT;
5+
import static java.util.concurrent.TimeUnit.SECONDS;
6+
7+
import datadog.trace.api.WellKnownTags;
8+
import datadog.trace.common.metrics.AdversarialMetricsBenchmark.CountingHealthMetrics;
9+
import datadog.trace.core.CoreSpan;
10+
import de.thetaphi.forbiddenapis.SuppressForbidden;
11+
import java.util.Collections;
12+
import java.util.List;
13+
import java.util.concurrent.ThreadLocalRandom;
14+
import org.openjdk.jmh.annotations.Benchmark;
15+
import org.openjdk.jmh.annotations.BenchmarkMode;
16+
import org.openjdk.jmh.annotations.Fork;
17+
import org.openjdk.jmh.annotations.Measurement;
18+
import org.openjdk.jmh.annotations.Mode;
19+
import org.openjdk.jmh.annotations.OutputTimeUnit;
20+
import org.openjdk.jmh.annotations.Scope;
21+
import org.openjdk.jmh.annotations.Setup;
22+
import org.openjdk.jmh.annotations.State;
23+
import org.openjdk.jmh.annotations.TearDown;
24+
import org.openjdk.jmh.annotations.Threads;
25+
import org.openjdk.jmh.annotations.Warmup;
26+
import org.openjdk.jmh.infra.Blackhole;
27+
28+
/**
29+
* Cardinality-isolation companion to {@link AdversarialMetricsBenchmark}: only the {@code resource}
30+
* dimension varies; {@code service}, {@code operation}, and {@code peer.hostname} are pinned to
31+
* single values. Pairing this with the adversarial bench (all four dimensions high-cardinality) and
32+
* {@link HighCardinalityPeerMetricsBenchmark} (only peer-tag high-cardinality) lets you attribute
33+
* any throughput delta to a specific axis.
34+
*
35+
* <p>Same shape as the adversarial bench -- 8 producer threads, {@code 0xFFFFF} (~1M) distinct
36+
* resource values which exceeds the default {@code tracerMetricsMaxAggregates=2048}, so the LRU
37+
* cache evicts continuously. Random {@code error}/{@code topLevel}/duration to keep histogram load
38+
* comparable; only the cardinality profile changes.
39+
*/
40+
@State(Scope.Benchmark)
41+
@Warmup(iterations = 2, time = 15, timeUnit = SECONDS)
42+
@Measurement(iterations = 5, time = 15, timeUnit = SECONDS)
43+
@BenchmarkMode(Mode.Throughput)
44+
@OutputTimeUnit(SECONDS)
45+
@Threads(8)
46+
@Fork(value = 1)
47+
public class HighCardinalityResourceMetricsBenchmark {
48+
49+
private ConflatingMetricsAggregator aggregator;
50+
private CountingHealthMetrics health;
51+
52+
@State(Scope.Thread)
53+
public static class ThreadState {
54+
int cursor;
55+
}
56+
57+
@Setup
58+
public void setup() {
59+
this.health = new CountingHealthMetrics();
60+
this.aggregator =
61+
new ConflatingMetricsAggregator(
62+
new WellKnownTags("", "", "", "", "", ""),
63+
Collections.emptySet(),
64+
new ConflatingMetricsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
65+
Collections.singleton("peer.hostname"), Collections.emptySet()),
66+
this.health,
67+
new ConflatingMetricsAggregatorBenchmark.NullSink(),
68+
2048,
69+
2048,
70+
false);
71+
this.aggregator.start();
72+
}
73+
74+
@TearDown
75+
@SuppressForbidden
76+
public void tearDown() {
77+
aggregator.close();
78+
System.err.println(
79+
"[HIGH_CARD_RESOURCE] drops over the trial (8 threads, warmup + measurement combined):");
80+
System.err.println(" onStatsInboxFull = " + health.inboxFull.sum());
81+
System.err.println(" onStatsAggregateDropped = " + health.aggregateDropped.sum());
82+
}
83+
84+
@Benchmark
85+
public void publish(ThreadState ts, Blackhole blackhole) {
86+
int idx = ts.cursor++;
87+
ThreadLocalRandom rng = ThreadLocalRandom.current();
88+
89+
int scrambled = idx * 0x9E3779B1;
90+
String resource = "res-" + ((scrambled ^ 0x5A5A5A) & 0xFFFFF);
91+
boolean error = (idx & 7) == 0;
92+
boolean topLevel = (idx & 3) == 0;
93+
long durationNanos = 1L + (rng.nextLong() & 0x3FFFFFFFL);
94+
95+
SimpleSpan span =
96+
new SimpleSpan("svc", "op", resource, "web", true, topLevel, error, 0, durationNanos, 200);
97+
span.setTag(SPAN_KIND, SPAN_KIND_CLIENT);
98+
span.setTag("peer.hostname", "localhost");
99+
100+
List<CoreSpan<?>> trace = Collections.singletonList(span);
101+
blackhole.consume(aggregator.publish(trace));
102+
}
103+
}

dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateMetric.java

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,27 @@ public AggregateMetric recordDurations(int count, AtomicLongArray durations) {
4646
return this;
4747
}
4848

49+
/**
50+
* Records a single hit. {@code tagAndDuration} carries the duration nanos with optional {@link
51+
* #ERROR_TAG} / {@link #TOP_LEVEL_TAG} bits OR-ed in.
52+
*/
53+
public AggregateMetric recordOneDuration(long tagAndDuration) {
54+
++hitCount;
55+
if ((tagAndDuration & TOP_LEVEL_TAG) == TOP_LEVEL_TAG) {
56+
tagAndDuration ^= TOP_LEVEL_TAG;
57+
++topLevelCount;
58+
}
59+
if ((tagAndDuration & ERROR_TAG) == ERROR_TAG) {
60+
tagAndDuration ^= ERROR_TAG;
61+
errorLatencies.accept(tagAndDuration);
62+
++errorCount;
63+
} else {
64+
okLatencies.accept(tagAndDuration);
65+
}
66+
duration += tagAndDuration;
67+
return this;
68+
}
69+
4970
public int getErrorCount() {
5071
return errorCount;
5172
}

0 commit comments

Comments
 (0)