Add span-derived primary tags (CSS v1.3.0) #11402
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| 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.SECONDS; | ||
|
|
||
| import datadog.trace.api.WellKnownTags; | ||
| import datadog.trace.core.CoreSpan; | ||
| import de.thetaphi.forbiddenapis.SuppressForbidden; | ||
| import java.util.Arrays; | ||
| import java.util.Collections; | ||
| import java.util.LinkedHashSet; | ||
| import java.util.List; | ||
| import java.util.concurrent.ThreadLocalRandom; | ||
| 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; | ||
|
|
||
| /** | ||
| * Regression benchmark for the additional-tags hot path; {@code limitsEnabled=true} is slower here | ||
| * because it records more (sentinel collapses), not because limiting is expensive. | ||
| */ | ||
| @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 = 1) | ||
| public class AdditionalTagsMetricsBenchmark { | ||
|
|
||
| private ClientStatsAggregator aggregator; | ||
| private AdversarialMetricsBenchmark.CountingHealthMetrics health; | ||
|
|
||
| @Param({"false", "true"}) | ||
| public boolean limitsEnabled; | ||
|
|
||
| @State(Scope.Thread) | ||
| public static class ThreadState { | ||
| int cursor; | ||
| } | ||
|
|
||
| @Setup | ||
| public void setup() { | ||
| this.health = new AdversarialMetricsBenchmark.CountingHealthMetrics(); | ||
| AdditionalTagsSchema additionalTagsSchema = | ||
| AdditionalTagsSchema.from( | ||
| new LinkedHashSet<>(Arrays.asList("region", "tenant_id")), | ||
| MetricCardinalityLimits.ADDITIONAL_TAG_VALUE, | ||
| limitsEnabled, | ||
| this.health); | ||
| this.aggregator = | ||
| new ClientStatsAggregator( | ||
| new WellKnownTags("", "", "", "", "", ""), | ||
| Collections.emptySet(), | ||
| additionalTagsSchema, | ||
| new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery( | ||
| Collections.singleton("peer.hostname"), Collections.emptySet()), | ||
| this.health, | ||
| new ClientStatsAggregatorBenchmark.NullSink(), | ||
| 2048, | ||
| 2048, | ||
| false); | ||
| this.aggregator.start(); | ||
| } | ||
|
|
||
| @TearDown | ||
| @SuppressForbidden | ||
| public void tearDown() { | ||
| aggregator.close(); | ||
| System.err.println("[ADDITIONAL-TAGS] counters (across all threads, single fork):"); | ||
| System.err.println(" onStatsInboxFull = " + health.inboxFull.sum()); | ||
| System.err.println(" onStatsAggregateDropped = " + health.aggregateDropped.sum()); | ||
| } | ||
|
|
||
| @Benchmark | ||
| public void publish(ThreadState ts, Blackhole blackhole) { | ||
| int idx = ts.cursor++; | ||
| ThreadLocalRandom rng = ThreadLocalRandom.current(); | ||
|
|
||
| int scrambled = idx * 0x9E3779B1; | ||
| String region = "region-" + ((scrambled >>> 4) & 0xFFFF); | ||
| String tenant = "tenant-" + ((scrambled >>> 16) & 0xFFFF); | ||
| long durationNanos = 1L + (rng.nextLong() & 0x3FFFFFFFL); | ||
|
|
||
| SimpleSpan span = | ||
| new SimpleSpan("svc", "op", "res", "web", true, true, false, 0, durationNanos, 200); | ||
| span.setTag(SPAN_KIND, SPAN_KIND_CLIENT); | ||
| span.setTag("region", region); | ||
| span.setTag("tenant_id", tenant); | ||
|
|
||
| List<CoreSpan<?>> trace = Collections.singletonList(span); | ||
| blackhole.consume(aggregator.publish(trace)); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| package datadog.trace.common.metrics; | ||
|
|
||
| import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; | ||
| import datadog.trace.core.monitor.HealthMetrics; | ||
| import java.util.ArrayList; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Immutable schema of configured span-derived primary tag keys; built once at aggregator | ||
| * construction. | ||
| */ | ||
| final class AdditionalTagsSchema { | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(AdditionalTagsSchema.class); | ||
|
|
||
| // Backend stats pipeline only supports a few primary tag dimensions; drop overflow at startup. | ||
| static final int MAX_ADDITIONAL_TAG_KEYS = 4; | ||
|
|
||
| /** Singleton empty schema returned when no additional tags are configured. */ | ||
| static final AdditionalTagsSchema EMPTY = | ||
| new AdditionalTagsSchema(new String[0], new TagCardinalityHandler[0], HealthMetrics.NO_OP); | ||
|
|
||
| final String[] names; | ||
|
|
||
| /** Per-key handlers providing UTF8 caching and per-cycle cardinality limiting. */ | ||
| private final TagCardinalityHandler[] handlers; | ||
|
|
||
| private final HealthMetrics healthMetrics; | ||
|
|
||
| private AdditionalTagsSchema( | ||
| String[] names, TagCardinalityHandler[] handlers, HealthMetrics healthMetrics) { | ||
| this.names = names; | ||
| this.handlers = handlers; | ||
| this.healthMetrics = healthMetrics; | ||
| } | ||
|
|
||
| /** Test convenience: uses {@link HealthMetrics#NO_OP} and limits enabled. */ | ||
| static AdditionalTagsSchema from(Set<String> configured) { | ||
| return from( | ||
| configured, MetricCardinalityLimits.ADDITIONAL_TAG_VALUE, true, HealthMetrics.NO_OP); | ||
| } | ||
|
|
||
| static AdditionalTagsSchema from( | ||
| Set<String> configured, int limit, boolean useBlockedSentinel, HealthMetrics healthMetrics) { | ||
| if (configured == null || configured.isEmpty()) { | ||
| return EMPTY; | ||
| } | ||
| List<String> valid = new ArrayList<>(); | ||
| for (String key : configured) { | ||
| if (key == null || key.isEmpty()) { | ||
| log.warn("Ignoring empty additional metric tag key"); | ||
| continue; | ||
| } | ||
| if (key.contains(":")) { | ||
| log.warn("Ignoring additional metric tag key '{}': keys must not contain ':'", key); | ||
| continue; | ||
| } | ||
| valid.add(key); | ||
| } | ||
| if (valid.isEmpty()) { | ||
| return EMPTY; | ||
| } | ||
| Collections.sort(valid); | ||
| // Dedup (sort brings duplicates adjacent) | ||
| List<String> deduped = new ArrayList<>(valid.size()); | ||
| String prev = null; | ||
| for (String key : valid) { | ||
| if (!key.equals(prev)) { | ||
| deduped.add(key); | ||
| prev = key; | ||
| } | ||
| } | ||
| if (deduped.size() > MAX_ADDITIONAL_TAG_KEYS) { | ||
| log.warn( | ||
| "Configured additional metric tag keys ({}) exceeds the supported limit of {}; " | ||
| + "dropping extra keys: {}", | ||
| deduped.size(), | ||
| MAX_ADDITIONAL_TAG_KEYS, | ||
| deduped.subList(MAX_ADDITIONAL_TAG_KEYS, deduped.size())); | ||
| deduped = deduped.subList(0, MAX_ADDITIONAL_TAG_KEYS); | ||
| } | ||
| String[] namesArr = deduped.toArray(new String[0]); | ||
| TagCardinalityHandler[] handlersArr = new TagCardinalityHandler[namesArr.length]; | ||
| for (int i = 0; i < namesArr.length; i++) { | ||
| handlersArr[i] = | ||
| new TagCardinalityHandler( | ||
| namesArr[i], | ||
| limit, | ||
| useBlockedSentinel, | ||
| MetricCardinalityLimits.ADDITIONAL_TAG_MAX_VALUE_LENGTH); | ||
| } | ||
| return new AdditionalTagsSchema(namesArr, handlersArr, healthMetrics); | ||
| } | ||
|
|
||
| int size() { | ||
| return names.length; | ||
| } | ||
|
|
||
| String name(int i) { | ||
| return names[i]; | ||
| } | ||
|
|
||
| UTF8BytesString register(int i, String value) { | ||
| return handlers[i].register(value); | ||
| } | ||
|
|
||
| void resetHandlers() { | ||
|
Contributor
There was a problem hiding this comment. Why is this schema class designed to have a It's primarily a design question - not recommending a better design here. I want to understand the rationale behind it or whether I'm missing an important design piece 🙇 . |
||
| for (int i = 0; i < handlers.length; i++) { | ||
| long blocked = handlers[i].reset(); | ||
| if (blocked > 0) { | ||
| log.warn( | ||
| "Cardinality limit reached for additional metric tag '{}'; further values will be reported as tracer_blocked_value", | ||
| names[i]); | ||
| healthMetrics.onTagCardinalityBlocked(handlers[i].statsDTag(), blocked); | ||
| } | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Curious, why do we choose to run benchmarks via annotations instead of programmatically via
OptionsBuilder?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Mostly, that's just what we've done historically.
Although in other places, I'm pushing for minimum number of threads & forks, so allocation translates into an obvious throughput hit. That was missed in a lot of places previously.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Makes sense to me - is this work we have recorded / planned somewhere? If not, we can create a ticket and expand on it later
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
My question came from a re-usability standpoint in case we plan to pass configuration there unless there is a way to read config with annotations