Update client-side stats to use light weight Hashtable

[dougqh]

What Does This Do

Replaces the MetricKey based HashMap with a new AggregateTable based on the light Hashtable

Motivation

By using the light Hashtable, I'm able to avoid the biggest source of allocation in client-side stats: MetricKey

Hashtable provides utilities for searching the entries without constructing a new composite key object

First, the components are hashed together to find the corresponding bucket
Then the bucket can be traversed to see if the entries match the key components

And the custom Entry can hold multiple fields that comprise the data / metadata needed for metric aggregation and eviction policy

The end result is that both MetricKey and AggregateMetric can be merged into a single class AggregateEntry that is only constructed when there's no existing matching entry

Additional Notes

Stacked on top of #11381 (dougqh/conflating-metrics-background-work). #11409 — the Hashtable utility this PR builds on — landed in master separately, so the diff shown here is only the AggregateTable / AggregateEntry / ClearSignal work plus follow-up cleanups.

Restructures the consumer-side aggregate store. Three logical commits, intended to be reviewed in order:

1. Add AggregateTable + AggregateEntry backed by Hashtable

Introduces a multi-key hash table that lets the consumer thread look up the {labels → counters} entry directly from a SpanSnapshot's raw fields — no MetricKey allocation per snapshot, no per-snapshot UTF8 cache lookups, no CHM operations. Hot-path lookup is keyHash compute → Hashtable.Support.bucket → bucket walk → matches(keyHash, snapshot) → returned entry has the counters to mutate in place.

This commit is standalone — no call sites yet, only the new classes + unit tests for hit/miss/cap-overrun/expunge/clear behavior.

2. Swap Aggregator to use AggregateTable + route disable() clear through a ClearSignal

Replaces LRUCache<MetricKey, AggregateMetric> with AggregateTable in Aggregator. Drops the AggregateExpiry listener — drop reporting (onStatsAggregateDropped) moves to the cap-overrun path inside Drainer.accept.

Threading fix bundled here: ConflatingMetricsAggregator.disable() used to call aggregator.clearAggregates() and inbox.clear() directly from the Sink's IO callback thread, racing with the aggregator thread. That race was tolerable for LinkedHashMap (worst case = corrupted internal state right before everything got cleared anyway); it's not tolerable for Hashtable (chain corruption can NPE or loop). disable() now offers a ClearSignal to the inbox so the aggregator thread itself performs the clear — preserves the single-writer invariant for AggregateTable end-to-end. The offer is best-effort; the system self-heals on a subsequent downgrade cycle if the inbox happens to be full (commented at the call site).

Cap-overrun semantic change: the old LRUCache evicted least-recently-used in O(1). AggregateTable instead scans for a hitCount==0 entry to recycle (O(N) worst-case), and drops the new key if none exists. Practical impact: in steady state, an unrelated burst of new keys gets dropped (and reported via onStatsAggregateDropped) rather than evicting established keys. The cost trade-off is commented at the eviction site — eviction is expected rare because the cap is sized to the working set; cursor-caching is the future option if a workload runs persistently at cap. The existing test that asserted "service0 evicted in favor of service10" is updated to assert the new semantics; the other cap-related test ("evicted entry was already flushed") still passes unchanged.

3. Fold MetricKey + AggregateMetric into AggregateEntry

MetricKey existed for two reasons — being the LRUCache key (replaced by AggregateTable's Hashtable mechanics) and being the labels arg to MetricWriter.add (the only thing left). AggregateMetric was the counter/histogram counterpart. Folds both onto a single AggregateEntry (10 UTF8 label fields + 3 primitives + counters + histograms), changes MetricWriter.add(MetricKey, AggregateMetric) → add(AggregateEntry), and deletes MetricKey.java + MetricKeys.java + AggregateMetric.java.

The 12 UTF8 caches that used to be split between MetricKey (9) and ConflatingMetricsAggregator (3, with overlap) are consolidated on AggregateEntry. One cache per field type now.

Latent bug fix: the prior matches(SpanSnapshot) used Objects.equals on raw fields. If the same logical key was delivered once as String and once as UTF8BytesString (different CharSequence impls of identical content), Objects.equals returned false and the table would split into two entries for the same key. The new matches uses content-equality (UTF8BytesString.toString() returns the underlying String in O(1)), collapsing them correctly.

Test impact: AggregateEntry.of(...) mirrors the prior new MetricKey(...) positional args, so test diffs are mostly mechanical. About 56 test sites migrated across ConflatingMetricAggregatorTest, SerializingMetricWriterTest, and MetricsIntegrationTest.

Review polish

Follow-up commits address review feedback:

• Use Hashtable.Support.create(maxAggregates, Support.MAX_RATIO) + Support.bucket + Support.insertHeadEntry(buckets, keyHash, entry) + Support.mutatingTableIterator to delegate to the helpers added on Add Hashtable and LongHashingUtils utilities #11409 — drops ~50 lines of bespoke bucket-array code.
• Inline the report-time forEach lambda instead of a static BiConsumer constant (the JIT reuses non-capturing lambdas).
• AggregateEntry.matches(long keyHash, SpanSnapshot) overload that pre-checks the hash, so chain walks read as one call.
• @Nullable (javax.annotation) annotations on the four nullable label fields + their getters + of(...) parameters.
• Objects.equals import in AggregateEntry.equals() (no more fully-qualified refs).
• Design-trade-off comments on evictOneStale (O(N) scan rationale) and disable() (best-effort offer rationale).

Additional cleanups

A second round of review surfacing landed these:

• MetricsIntegrationTest .aggregate runtime bug — the legacy entry.aggregate.recordDurations(...) form compiled under Groovy's dynamic dispatch but would have thrown MissingPropertyException at runtime. Fixed to call recordDurations directly on the entry. (Bot-flagged.)
• Spock >> { closure } no-op assertions in ConflatingMetricAggregatorTest — the >> operator stubs a return value, so the closures verifying e.getHitCount() == X && ... were being evaluated and discarded. Wrapped 31 sites in assert so Groovy power-assert surfaces mismatches. All 41 tests still pass, so the previously-unverified assertions happened to hold. (Bot-flagged.)
• Drop dead recordDurations(int, AtomicLongArray) batch API — vestige of master's Batch design. Production now only calls recordOneDuration(long). Migrated the three remaining test callers (AggregateEntryTest, SerializingMetricWriterTest, MetricsIntegrationTest) to loops of recordOneDuration calls, then deleted the batched method and its AtomicLongArray imports.
• AggregateEntry.of() colon-split Javadoc warning — the test factory recovers (name, value) pairs from "name:value" strings by splitting at the first :, which is brittle if a peer-tag value contains a colon (URLs, IPv6, service:env). Added an explicit warning so callers know to keep test data colon-free in values.
• ConflatingMetricsAggregatorDisableTest — new JUnit 5 coverage for the disable() → ClearSignal threading routing. The test fires DOWNGRADED from the test thread, waits for the no-flush window, then publishes a marker span with a distinct resource name and asserts the next flush captures only the marker — proving CLEAR actually wiped the original entry from the table. Catches both the missing-clear regression and the bucket-chain-corruption regression that the original threading race could produce.

Benchmarks

Producer publish() latency (single-threaded, 2 forks × 5 iter × 15s)

	Prior commit (stacked base)	This PR
SimpleSpan bench	3.116 µs/op	3.123 µs/op
DDSpan bench	2.506 µs/op	2.412 µs/op

All within noise on the producer side — this PR is a consumer-side refactor, so producer publish() shouldn't move much. The structural wins (one less class, no per-snapshot MetricKey allocation on the consumer, no double-cache lookups, smaller per-entry footprint) only become visible when the consumer is hammered hard enough that snapshot processing rate matters. That's exactly what the adversarial bench measures.

AdversarialMetricsBenchmark (8 producer threads, 2×15s warmup + 5×15s, 1 fork)

Same benchmark used on #11381 (high-cardinality (service, operation, resource, peer.hostname) per op, random durations across 1ns–1s, random error/topLevel flags — designed to saturate every capacity bound at once).

	master	#11381 (parent)	this PR
Throughput avg (ops/s)	395,806 ± 2,619,133	4,889,660 ± 390,175	27,915,800 ± 1,219,470
Per-iteration progression (ops/s)	warmup 2,536,145 → 205,314 → 95,888 → 47,301 → 24,378	4,886,778 → 4,875,195 → 4,731,827 → 4,959,992 → 4,994,511	28,043,909 → 28,112,828 → 27,354,721 → 28,074,395 → 27,993,147
Stdev (ops/s)	680,180	101,327	316,692
onStatsInboxFull (drops at handoff)	n/a (no inbox on master)	199,862,634	2,893,855,052
onStatsAggregateDropped	11,642,039	84,002,323	16,301,696

~70× faster than master, ~5.7× faster than #11381 alone. The producer fast-path drop check (inbox.size() >= inbox.capacity()) returns immediately when the inbox is saturated, so when the consumer can drain ~5× faster (this PR's consumer-side win), the producer spins through publish() at the fast-path rate and inbox-full drops climb correspondingly (200M on #11381 → 2.9 B here). That's the design working as intended: under attack, load shedding happens at the cheapest place in the pipeline.

The aggregate-cache drop count actually fell (84M → 16M) because the faster consumer keeps the table from staying at cap as often — fewer snapshots arrive at a cap-overrun state with no stale entry to recycle.

Per-iteration shape stays flat (27.4M–28.1M ops/s, no warmup → measurement degradation), confirming the design holds steady-state under sustained load.

Cardinality-isolation companions (8 producer threads, 2×15s warmup + 5×15s, 1 fork)

HighCardinalityResourceMetricsBenchmark and HighCardinalityPeerMetricsBenchmark (added in #11381) pin every dimension except one to attribute throughput deltas to a specific axis.

HighCardinalityResourceMetricsBenchmark — only resource varies (~1M distinct), service/operation/peer.hostname pinned:

	master	#11381 (parent)	this PR
Throughput avg (ops/s)	5,089,853 ± 587,996	8,983,726 ± 2,677,454	28,173,068 ± 485,923
Per-iter ops/s (5 measurement)	5,277K → 5,014K → 4,948K → 4,978K → 5,232K	8,043K → 8,844K → 9,962K → 9,217K → 8,852K	28,147K → 28,391K → 28,094K → 28,079K → 28,155K
onStatsInboxFull	0 (no inbox)	611,612,206	2,904,235,942
onStatsAggregateDropped	12,379,590	325,582,103	17,033,533

HighCardinalityPeerMetricsBenchmark — only peer.hostname varies (~32K distinct), service/operation/resource pinned:

	master	#11381 (parent)	this PR
Throughput avg (ops/s)	6,335,821 ± 1,068,632	9,255,788 ± 2,933,935	30,651,802 ± 1,108,285
Per-iter ops/s (5 measurement)	6,552K → 6,510K → 6,429K → 6,323K → 5,865K	9,137K → 10,258K → 9,752K → 8,794K → 8,337K	30,876K → 30,722K → 30,852K → 30,644K → 30,166K
onStatsInboxFull	0	744,795,870	3,107,668,314
onStatsAggregateDropped	7,039,180	208,556,622	17,532,646

+454 % over master / +214 % over #11381 on the resource axis; +384 % / +231 % on the peer axis. The pattern is consistent with the adversarial run: master is sensitive to which dimension is wide (resource at ~1M hits pending/keys CHM contention harder than peer.hostname at ~32K); this PR runs at ~28–30 M ops/s on either bench because the consumer is no longer the bottleneck. Variance also collapses (±486K / ±1.1M here vs ±2.7M / ±2.9M on #11381), reflecting that the new hashtable hits steady-state immediately with no JIT warmup tail.

The drop-counter shape on this PR flips even further toward inbox-full (170× ratio against aggregate-dropped on both benches), confirming that the consumer drains so fast the LRU aggregate cache barely needs to evict.

Net code delta: +1280 / −903 = +377 lines across 16 files. The growth is dominated by new test coverage (AggregateTableTest, AggregateEntryTest) plus the consolidated UTF8 caches landing on AggregateEntry; the production-code core (less MetricKey + MetricKeys + AggregateMetric minus AggregateEntry's additions) is roughly flat.

Test plan

• ./gradlew :dd-trace-core:test --tests 'datadog.trace.common.metrics.*' passes (incl. the new AggregateTableTest and AggregateEntryTest)
• ./gradlew :dd-trace-core:compileJava :dd-trace-core:compileTestGroovy :dd-trace-core:compileJmhJava :dd-trace-core:compileTraceAgentTestGroovy all green
• ./gradlew spotlessCheck clean
• CI muzzle / integration suites
• Validate stats.dropped_aggregates semantics at high cardinality (especially the new "drop new on cap overrun" path vs. the old "evict LRU" path)

🤖 Generated with Claude Code

[datadog-datadog-prod-us1-2]

✨ Fix all issues with BitsAI

⚠️ Warnings

🚦 2 Pipeline jobs failed

DataDog/apm-reliability/dd-trace-java | agent_integration_tests     

🔧 Fix in code (Fix with Cursor).

4 failed tests due to java.lang.IllegalAccessError at MetricsIntegrationTest.groovy:45.

DataDog/apm-reliability/dd-trace-java | check_base     

🔧 Fix in code (Fix with Cursor).

CodeNarc rule violations found. Exceeded maximum number of priority 3 violations during task ':dd-trace-core:codenarcTraceAgentTest'.

Useful? React with 👍 / 👎

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 4a4b72e | Docs | Datadog PR Page | Give us feedback!}

[amarziali]

before the service name was also set to EMPTY when null. I think this in reality cannot happen but perhaps it's worth going through canonicalize also for this

[dougqh]

Yes, I need to revisit the null vs EMPTY story. As I recall the null handling was already a mix of two approaches before, but we should fix it.

[dougqh]

From Claude (posted on dougqh's behalf):

Done in d1749389b9. service and spanKind now both go through canonicalize like every other label field — null maps to EMPTY instead of NPE'ing at the cache call. Dropped the Functions.UTF8_ENCODE import too since the canonicalize helper handles the create.

Pinned by AggregateTableTest.nullServiceAndSpanKindDoNotNpeAndCollapseWithEmpty which publishes (null, null), (null, null) again, and ("", ""), asserts a single entry and length-0 service/spanKind. Without the fix, the first publish would NPE'd at .computeIfAbsent.

[amarziali]

this looks now a dead method. Given that is not thread safe wrt the underlying hashtable I would remove it or reduce its visibility

[dougqh]

From Claude (posted on dougqh's behalf):

Deleted in 21e75452d6 — the method had no callers after the disable() rewrite, and you're right that exposing a thread-unsafe mutator from outside the inbox routing would invite the exact race that motivated the ClearSignal path. The comment in ConflatingMetricsAggregator that named it has been updated to describe the inbox-routed clear instead.

[amarziali]

Minor: this null check makes the equals not symmetrical since we're skipping here the field from including into the hash - also combinations with different nulls on peerTagSchema/peerTagValues can produce the same hash theoretically

[dougqh]

I think at this point I should just get rid of equals & hashCode. They aren't really used by the new table structure.

As for the nulls in the hash code, yes, that can lead to hash collisions, but that's fine.
That said, I do think I'll rework the hashOf routine to be a bit lighter weight. I may or may not change the null handling when I do so.

[dougqh]

From Claude (posted on dougqh's behalf):

Both halves of this are now fixed:

• Hash side (80778c4e87): dropped the conditional null-skip. hashOf now unconditionally mixes in LongHashingUtils.addToHash(h, s.peerTagSchema) and Arrays.hashCode(s.peerTagValues). PeerTagSchema.hashCode() is content-based on names (lazily cached), so distinct schema instances with identical names still produce a stable hash. Different-null combinations on the two fields can no longer collide.

• Equals side (2536aa2e76): AggregateEntry.equals now compares the raw peerTagNames / peerTagValues arrays via Arrays.equals, mirroring exactly what hashOf hashes. Same for be13431744 on PeerTagSchema (symmetric equals/hashCode pair on names).

Two regression tests pin the contract: AggregateEntryTest.equalsConsistentWithHashCodeAcrossDifferentSchemaLayouts (entries with ["a","b"]/[null,"x"] vs ["b","c"]/["x",null] — same encoded peer-tags ["b:x"] but different raw layouts → not equal, different hashCodes) and equalEntriesHaveEqualHashCodes (the positive case).

[amarziali]

What highlighted in the comment is true. However if I'm not wrong when the aggregator start dropping (meaning that capacity is full and badly dimensioned) this will translate in a perf kill. I'm wondering if it's worth adding a field for the last scanned bucket to have the cost amortized right now

[dougqh]

Can you clarify "perf kill"? Yes, this will end up doing an O(n) scan of the table, but the table is small.
Yes, that will slow down the metrics thread, but it won't kill the application.

I've primarily focused on making sure things don't fall over under an adversarial workload. In those cases, such last used caches don't work. Although, I suppose I could check a more realistic accidental case where only a single component has out of control cardinality.

[amarziali]

was about needing to scan the full table for each span inserted that at the end cannot insert since will overflow. but perhaps I've misinterpret something

[dougqh]

Okay, I'll double check the eviction logic. That is worth being careful about.
Although, the next part that introduces cardinality limiters largely renders the table limit unnecessary.

[dougqh]

I updated the eviction to track the last bucket from one scan to the next.
To do that, I added a variation of the Hashtable iterator that scans a range of buckets.

Since only this code is using Hashtable right now, I went ahead and included that in this PR

[amarziali]

should the dirty flag be also cleared up here?

[dougqh]

Good question. I think I need to go back and review all the signaling a bit more.
There were some latent problems that already existed in the code that have been exposed during the reviews.

[dougqh]

Added in the clearing of the dirty flag.

I wonder if that was a latent bug because I mostly intended just to shuffle around the existing code. It might have been that we were just getting a spurious reporting cycle previously.

Anyway, it is fixed now.

[sarahchen6]

can remove this method and just use contentEquals above since String is a CharSequence?

[sarahchen6]

nit: this method could be simplified to something like

private static boolean contentEquals(UTF8BytesString a, CharSequence b) {
  if (a == null || a.length() == 0) {
    return b == null || b.length() == 0;
  }
  return b != null && a.toString().contentEquals(b);
}

[sarahchen6]

since this is now test-specific, it would make sense to me to have a more test-related name, instead of reading nearly identical to AggregateEntry?

[sarahchen6]

nit: I think this comment could be removed. It seems to explain a fix implemented during the PR iteration process and how this test case addresses that fix, but the test case can stand on its own, i.e. future code readers only need to know what the final code expected behavior is.

this may apply to the comment below too for nullServiceAndSpanKindDoNotNpeAndCollapseWithEmpty()

[sarahchen6]

same here... maybe keeping the comment as just:

CLEAR handler clears only the aggregates table; queued signals (STOP, REPORT) survive and get processed normally.

basically removing references to the PR iteration process and keeping just the expected behavior / objectively helpful context in comments

[sarahchen6]

nit: this line seems unnecessary given 100 and 102 🤔

[sarahchen6]

Left a few nits, but otherwise looks reasonable!!

Also this was Claude's advice for the failing agent integration test:

Fix options (in order of invasiveness):
1. Add a public test-support module — extract a small AggregateEntryTestSupport factory into a test-support
jar that traceAgentTest can depend on, keeping the production classes package-private.
2. Make testSchema + forSnapshot public on their classes — simplest, but widens the API surface of
intentionally-internal classes.
3. Rewrite the test to not reference these classes directly — drive the test entirely via
ConflatingMetricsAggregator publishing real DDSpans, which avoids any dependency on the internal
AggregateEntry/PeerTagSchema types.

Option 3 is the cleanest because it also makes the integration test more realistic (it exercises the full
pipeline instead of bypassing the producer path).