From 6c0e91c49355e34077095a1dbfd67b607d0f9916 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 30 Oct 2026 14:52:23 -0400 Subject: [PATCH 1/6] Extract peer connection tags with a TagExtractor Move BaseDecorator.onPeerConnection's InetAddress / InetSocketAddress handling into PeerAddress and PeerSocketAddress extractors, applied with setTagsFrom so the span's tag lock is taken once per call. This is the first step toward freezing the extracted tags once per long-lived address instead of rebuilding them, and the new IP string, on every span. Co-Authored-By: Claude Opus 5.5 --- .../decorator/BaseDecorator.java | 21 +--- .../decorator/PeerAddress.java | 43 +++++++++ .../decorator/PeerSocketAddress.java | 30 ++++++ .../decorator/PeerAddressTest.java | 96 +++++++++++++++++++ 4 files changed, 171 insertions(+), 19 deletions(-) create mode 100644 dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java create mode 100644 dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerSocketAddress.java create mode 100644 dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddressTest.java diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java index 55a8784aee3..e889750638e 100644 --- a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java @@ -1,7 +1,5 @@ package datadog.trace.bootstrap.instrumentation.decorator; -import static datadog.trace.bootstrap.instrumentation.java.net.HostNameResolver.hostName; - import datadog.appsec.api.blocking.BlockingException; import datadog.context.Context; import datadog.context.ContextScope; @@ -15,8 +13,6 @@ import datadog.trace.bootstrap.instrumentation.api.ErrorPriorities; import datadog.trace.bootstrap.instrumentation.api.SpanPrototype; import java.lang.reflect.Method; -import java.net.Inet4Address; -import java.net.Inet6Address; import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.concurrent.ExecutionException; @@ -196,10 +192,7 @@ protected void doOnError(final AgentSpan span, final Throwable throwable, byte e public final void onPeerConnection( final AgentSpan span, @Nullable final InetSocketAddress remoteConnection) { - if (remoteConnection != null) { - onPeerConnection(span, remoteConnection.getAddress(), !remoteConnection.isUnresolved()); - setPeerPort(span, remoteConnection.getPort()); - } + span.setTagsFrom(remoteConnection, PeerSocketAddress.INSTANCE); } public final void onPeerConnection( @@ -209,17 +202,7 @@ public final void onPeerConnection( public final void onPeerConnection( AgentSpan span, @Nullable InetAddress remoteAddress, boolean resolved) { - if (remoteAddress != null) { - String ip = remoteAddress.getHostAddress(); - if (resolved && Config.get().isPeerHostNameEnabled()) { - span.setTag(KnownTags.PEER_HOSTNAME_ID, hostName(remoteAddress, ip)); - } - if (remoteAddress instanceof Inet4Address) { - span.setTag(KnownTags.PEER_IPV4_ID, ip); - } else if (remoteAddress instanceof Inet6Address) { - span.setTag(KnownTags.PEER_IPV6_ID, ip); - } - } + span.setTagsFrom(remoteAddress, resolved ? PeerAddress.RESOLVED : PeerAddress.UNRESOLVED); } public void setPeerPort(AgentSpan span, String port) { diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java new file mode 100644 index 00000000000..7dac17ab4a6 --- /dev/null +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java @@ -0,0 +1,43 @@ +package datadog.trace.bootstrap.instrumentation.decorator; + +import static datadog.trace.bootstrap.instrumentation.java.net.HostNameResolver.hostName; + +import datadog.trace.api.Config; +import datadog.trace.api.KnownTags; +import datadog.trace.bootstrap.instrumentation.api.SpanData; +import datadog.trace.bootstrap.instrumentation.api.TagExtractor; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.net.InetAddress; + +/** + * Extracts the peer tags of a remote {@link InetAddress}: its IP, as {@code peer.ipv4} or {@code + * peer.ipv6}, and, when the address was resolved and peer hostnames are enabled, {@code + * peer.hostname}. + */ +public final class PeerAddress implements TagExtractor { + /** For an address that was resolved from a hostname. */ + public static final PeerAddress RESOLVED = new PeerAddress(true); + + /** For an address that wasn't resolved: no {@code peer.hostname}. */ + public static final PeerAddress UNRESOLVED = new PeerAddress(false); + + private final boolean resolved; + + private PeerAddress(boolean resolved) { + this.resolved = resolved; + } + + @Override + public void extract(InetAddress address, SpanData span) { + String ip = address.getHostAddress(); + if (resolved && Config.get().isPeerHostNameEnabled()) { + span.setTag(KnownTags.PEER_HOSTNAME_ID, hostName(address, ip)); + } + if (address instanceof Inet4Address) { + span.setTag(KnownTags.PEER_IPV4_ID, ip); + } else if (address instanceof Inet6Address) { + span.setTag(KnownTags.PEER_IPV6_ID, ip); + } + } +} diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerSocketAddress.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerSocketAddress.java new file mode 100644 index 00000000000..3a77831775d --- /dev/null +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerSocketAddress.java @@ -0,0 +1,30 @@ +package datadog.trace.bootstrap.instrumentation.decorator; + +import datadog.trace.api.KnownTags; +import datadog.trace.bootstrap.instrumentation.api.SpanData; +import datadog.trace.bootstrap.instrumentation.api.TagExtractor; +import java.net.InetAddress; +import java.net.InetSocketAddress; + +/** + * Extracts the peer tags of a remote {@link InetSocketAddress}: those of its address (see {@link + * PeerAddress}), when it has one, and {@code peer.port}, when set. + */ +public final class PeerSocketAddress implements TagExtractor { + public static final PeerSocketAddress INSTANCE = new PeerSocketAddress(); + + private PeerSocketAddress() {} + + @Override + public void extract(InetSocketAddress socketAddress, SpanData span) { + // an unresolved socket address is exactly one with no address + InetAddress address = socketAddress.getAddress(); + if (address != null) { + PeerAddress.RESOLVED.extract(address, span); + } + int port = socketAddress.getPort(); + if (port > BaseDecorator.UNSET_PORT) { + span.setTag(KnownTags.PEER_PORT_ID, port); + } + } +} diff --git a/dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddressTest.java b/dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddressTest.java new file mode 100644 index 00000000000..a0083183716 --- /dev/null +++ b/dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddressTest.java @@ -0,0 +1,96 @@ +package datadog.trace.bootstrap.instrumentation.decorator; + +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import datadog.trace.api.KnownTags; +import datadog.trace.bootstrap.instrumentation.api.SpanData; +import datadog.trace.test.junit.utils.config.WithConfig; +import datadog.trace.test.junit.utils.config.WithConfigExtension; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +// registered for the class so the one method-level @WithConfig is undone after that method +@ExtendWith(WithConfigExtension.class) +class PeerAddressTest { + private final SpanData span = mock(SpanData.class); + + // getByAddress pairs a hostname with an IP without a DNS lookup + private static InetAddress ipv4() throws Exception { + return InetAddress.getByAddress("db.example", new byte[] {10, 0, 0, 1}); + } + + private static InetAddress ipv6() throws Exception { + byte[] ip = new byte[16]; + ip[15] = 1; + return InetAddress.getByAddress("db6.example", ip); + } + + @Test + void resolvedIpv4SetsHostnameAndIpv4() throws Exception { + PeerAddress.RESOLVED.extract(ipv4(), span); + + verify(span).setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) "db.example"); + verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); + verifyNoMoreInteractions(span); + } + + @Test + void resolvedIpv6SetsHostnameAndIpv6() throws Exception { + InetAddress address = ipv6(); + PeerAddress.RESOLVED.extract(address, span); + + verify(span).setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) "db6.example"); + verify(span).setTag(KnownTags.PEER_IPV6_ID, (CharSequence) address.getHostAddress()); + verifyNoMoreInteractions(span); + } + + @Test + void unresolvedSkipsHostname() throws Exception { + PeerAddress.UNRESOLVED.extract(ipv4(), span); + + verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); + verifyNoMoreInteractions(span); + } + + @Test + @WithConfig(key = "trace.peer.hostname.enabled", value = "false") + void hostnameDisabledSkipsHostname() throws Exception { + PeerAddress.RESOLVED.extract(ipv4(), span); + + verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); + verifyNoMoreInteractions(span); + } + + @Test + void socketAddressAddsPort() throws Exception { + PeerSocketAddress.INSTANCE.extract(new InetSocketAddress(ipv4(), 5432), span); + + verify(span).setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) "db.example"); + verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); + verify(span).setTag(KnownTags.PEER_PORT_ID, 5432); + verifyNoMoreInteractions(span); + } + + @Test + void unresolvedSocketAddressSetsOnlyPort() { + PeerSocketAddress.INSTANCE.extract( + InetSocketAddress.createUnresolved("bad.address.local", 999), span); + + verify(span).setTag(KnownTags.PEER_PORT_ID, 999); + verifyNoMoreInteractions(span); + } + + @Test + void unsetPortIsSkipped() throws Exception { + PeerSocketAddress.INSTANCE.extract(new InetSocketAddress(ipv4(), 0), span); + + verify(span, never()).setTag(eq(KnownTags.PEER_PORT_ID), anyInt()); + } +} From 16ecf44d3d311a5c02ccbdffd981c5b9a2f25542 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 30 Oct 2026 15:01:05 -0400 Subject: [PATCH 2/6] Add FrozenTags, a cached form of setTagsFrom FrozenTags.of(source, extractor) records the tags an extractor or contributor writes, once, into a frozen TagMap. span.setTagsFrom(frozen) then applies them as if the writes were made again, without reading the source. On DDSpan this reuses the prototype seed plan: with the dense store, known tags that need no interception and aren't already on the span are copied in at once; otherwise they go one at a time through the interceptor, as setTag does. The recorder rejects reads of the span and span-level setters, since neither can be replayed against a different span. Co-Authored-By: Claude Opus 5.5 --- .../main/java/datadog/trace/core/DDSpan.java | 10 + .../datadog/trace/core/DDSpanContext.java | 60 ++++- .../core/DDSpanFrozenTagsDenseForkedTest.java | 103 +++++++++ .../trace/core/DDSpanFrozenTagsTest.java | 200 +++++++++++++++++ .../instrumentation/api/AgentSpan.java | 12 + .../instrumentation/api/FrozenTags.java | 207 ++++++++++++++++++ .../instrumentation/api/ImmutableSpan.java | 6 + .../instrumentation/api/FrozenTagsTest.java | 107 +++++++++ 8 files changed, 695 insertions(+), 10 deletions(-) create mode 100644 dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java create mode 100644 dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java create mode 100644 internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java create mode 100644 internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java diff --git a/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java b/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java index df38b71aed7..41bd614d1f3 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java @@ -35,6 +35,7 @@ import datadog.trace.bootstrap.instrumentation.api.AgentSpanLink; import datadog.trace.bootstrap.instrumentation.api.AttachableWrapper; import datadog.trace.bootstrap.instrumentation.api.ErrorPriorities; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; import datadog.trace.bootstrap.instrumentation.api.ResourceNamePriorities; import datadog.trace.bootstrap.instrumentation.api.SpanPrototype; import datadog.trace.bootstrap.instrumentation.api.SpanWrapper; @@ -615,6 +616,15 @@ public DDSpan setTagsFrom(final TagContributor contributor) { return this; } + /** Applies frozen tags under the span's tag lock, copying them in at once where it can. */ + @Override + public DDSpan setTagsFrom(final FrozenTags frozen) { + if (frozen != null && !frozen.isEmpty()) { + context.setTagsFrom(frozen); + } + return this; + } + @Override public AgentSpan setAllTags(Map map) { context.setAllTags(map); diff --git a/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java b/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java index e1fa409f13c..a73496a77e1 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java @@ -40,6 +40,7 @@ import datadog.trace.bootstrap.instrumentation.api.Baggage; import datadog.trace.bootstrap.instrumentation.api.ClientIpAddressData; import datadog.trace.bootstrap.instrumentation.api.ErrorPriorities; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; import datadog.trace.bootstrap.instrumentation.api.ProfilerContext; import datadog.trace.bootstrap.instrumentation.api.ProfilingContextIntegration; import datadog.trace.bootstrap.instrumentation.api.ResourceNamePriorities; @@ -758,8 +759,9 @@ public void applyOverwriting(@Nonnull final SpanPrototype prototype) { } /** - * How a {@link SpanPrototype}'s tags are seeded into a span, cached on the prototype for the - * interceptor it was checked against ({@link SpanPrototype#seedPlan}). + * How a {@link SpanPrototype}'s or {@link FrozenTags}' tags are copied into a span, cached on the + * prototype or frozen tags for the interceptor it was checked against ({@link + * SpanPrototype#seedPlan}, {@link FrozenTags#applyPlan}). */ private static final class SeedPlan { final TagInterceptor interceptor; @@ -783,20 +785,36 @@ private static final class SeedPlan { private SeedPlan seedPlanFor(final SpanPrototype prototype) { Object cached = prototype.seedPlan(); - if (cached instanceof SeedPlan && ((SeedPlan) cached).interceptor == tagInterceptor) { + if (isCurrentPlan(cached)) { return (SeedPlan) cached; } - TagMap tags = prototype.tags(); - TagMap.Entry kind = tags.getEntry(KnownTags.SPAN_KIND_ID); - SeedPlan plan = - new SeedPlan( - tagInterceptor, - tags.knownStore().supportsBulkSeed() && tagInterceptor.canSeedInBulk(tags), - kind == null ? SPAN_KIND_UNSET : spanKindOrdinalOf(kind.stringValue())); + SeedPlan plan = newSeedPlan(prototype.tags()); prototype.seedPlan(plan); return plan; } + private SeedPlan seedPlanFor(final FrozenTags frozen) { + Object cached = frozen.applyPlan(); + if (isCurrentPlan(cached)) { + return (SeedPlan) cached; + } + SeedPlan plan = newSeedPlan(frozen.tags()); + frozen.applyPlan(plan); + return plan; + } + + private boolean isCurrentPlan(final Object cached) { + return cached instanceof SeedPlan && ((SeedPlan) cached).interceptor == tagInterceptor; + } + + private SeedPlan newSeedPlan(final TagMap tags) { + TagMap.Entry kind = tags.getEntry(KnownTags.SPAN_KIND_ID); + return new SeedPlan( + tagInterceptor, + tags.knownStore().supportsBulkSeed() && tagInterceptor.canSeedInBulk(tags), + kind == null ? SPAN_KIND_UNSET : spanKindOrdinalOf(kind.stringValue())); + } + /** * Seeds a prototype's tags where the span doesn't already have them, so explicit tags keep * precedence over the prototype's constant defaults. Two paths: @@ -1974,6 +1992,28 @@ void setTagsFrom(final TagContributor contributor) { } } + /** + * Applies frozen tags as if their recorded writes were made now, under one lock: in bulk when the + * plan allows it and the masks prove none of the known tags is already on the span ({@link + * #seedKnownInBulk}), otherwise one at a time through the interceptor, as {@code setTag} does. + * Custom tags always go one at a time. Returns whether the known tags went in bulk, for tests. + */ + boolean setTagsFrom(final FrozenTags frozen) { + final SeedPlan plan = seedPlanFor(frozen); + final TagMap tags = frozen.tags(); + synchronized (knownTags) { + if (seedKnownInBulk(plan, tags)) { + HashStore custom = tags.customStore(); + if (custom != null) { + custom.forEach(this, (ctx, tagEntry) -> ctx.putIntercepted(tagEntry)); + } + return true; + } + tags.forEach(this, (ctx, tagEntry) -> ctx.putIntercepted(tagEntry)); + return false; + } + } + /** Whether the current thread holds this span's tag lock; for assertions. */ boolean holdsTagLock() { return Thread.holdsLock(knownTags); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java new file mode 100644 index 00000000000..49c141984b7 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java @@ -0,0 +1,103 @@ +package datadog.trace.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.api.DDTags; +import datadog.trace.api.IdStore; +import datadog.trace.api.KnownTags; +import datadog.trace.api.config.TracerConfig; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; +import datadog.trace.bootstrap.instrumentation.api.TagContributor; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.test.junit.utils.config.WithConfig; +import org.junit.jupiter.api.Test; + +/** + * {@link DDSpanFrozenTagsTest} with the dense store on, where frozen known tags that need no + * interception and aren't already on the span are copied in at once ({@code + * DDSpanContext.setTagsFrom(FrozenTags)}); runs in its own JVM because the flag is read once. Adds + * which path each shape takes: the inherited tests pass on either, so these pin that the bulk copy + * is actually used. + */ +@WithConfig(key = TracerConfig.TRACE_DENSE_TAGS_ENABLED, value = "true") +class DDSpanFrozenTagsDenseForkedTest extends DDSpanFrozenTagsTest { + private static final FrozenTags PEER_TAGS = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(KnownTags.PEER_HOSTNAME_ID, "db.example"); + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + target.setTag(KnownTags.PEER_PORT_ID, 5432); + }); + + @Test + void theDenseStoreIsOn() { + // guards the rest of this class: with the flag missed, every test would quietly run on the + // hash store, where the frozen tag tests pass too + assertEquals("DenseIdStore", IdStore.create(null).getClass().getSimpleName()); + } + + @Test + void peerTagsOnAFreshSpanGoInBulk() { + DDSpan span = start(); + try { + assertTrue(span.spanContext().setTagsFrom(PEER_TAGS)); + + assertEquals("db.example", span.getTag(Tags.PEER_HOSTNAME)); + assertEquals("10.0.0.1", span.getTag(Tags.PEER_HOST_IPV4)); + assertEquals(5432, span.getTag(Tags.PEER_PORT)); + } finally { + span.finish(); + } + } + + @Test + void spanKindStillGoesInBulk() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CLIENT); + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); + DDSpan span = (DDSpan) tracer.buildSpan("test-instr", "frozen-tags.op").start(); + try { + assertTrue(span.spanContext().setTagsFrom(frozen)); + assertEquals(Tags.SPAN_KIND_CLIENT, span.getSpanKindString()); // the bulk path sets it + } finally { + span.finish(); + } + } + + @Test + void anInterceptedTagSendsTheSetOneAtATime() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(DDTags.RESOURCE_NAME, "frozen-resource"); + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); + DDSpan span = start(); + try { + assertFalse(span.spanContext().setTagsFrom(frozen)); + assertEquals("frozen-resource", span.getResourceName().toString()); + } finally { + span.finish(); + } + } + + @Test + void aTagTheSpanAlreadyHasSendsTheSetOneAtATime() { + DDSpan span = start(); + try { + span.setTag(Tags.PEER_PORT, 1); + assertFalse(span.spanContext().setTagsFrom(PEER_TAGS)); + assertEquals(5432, span.getTag(Tags.PEER_PORT)); // still overwritten + } finally { + span.finish(); + } + } +} diff --git a/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java new file mode 100644 index 00000000000..54bba7717ab --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java @@ -0,0 +1,200 @@ +package datadog.trace.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +import datadog.trace.api.DDTags; +import datadog.trace.api.KnownTags; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; +import datadog.trace.bootstrap.instrumentation.api.SpanData; +import datadog.trace.bootstrap.instrumentation.api.TagContributor; +import datadog.trace.bootstrap.instrumentation.api.TagExtractor; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.common.writer.ListWriter; +import java.util.TreeMap; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * {@code DDSpan.setTagsFrom(FrozenTags)}: applying recorded tags is equivalent to making the + * recorded writes on the span. {@link DDSpanFrozenTagsDenseForkedTest} reruns these with the dense + * store, where the known tags can be copied in at once. + */ +class DDSpanFrozenTagsTest extends DDCoreJavaSpecification { + CoreTracer tracer; + + @BeforeEach + void setup() { + tracer = tracerBuilder().writer(new ListWriter()).build(); + } + + @AfterEach + void tearDown() { + tracer.close(); + } + + DDSpan start() { + return (DDSpan) tracer.startSpan("test-instr", "frozen-tags.op"); + } + + /** + * A non-String value, shared so both spans hold the same object (StringBuilder has no equals). + */ + private static final Object OBJECT_VALUE = new StringBuilder("built"); + + /** Known tags by id and by name, custom tags, every value type, and a set-then-removed tag. */ + private static void writeTags(SpanData span) { + span.setTag(KnownTags.HTTP_ROUTE_ID, "/users/{id}"); + span.setTag(KnownTags.HTTP_STATUS_CODE_ID, 503); + span.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + span.setTag(Tags.PEER_PORT, 5432); + span.setTag("custom.string", "value"); + span.setTag("custom.object", OBJECT_VALUE); + span.setTag("custom.bool", true); + span.setTag("custom.long", 8L); + span.setTag("custom.double", 2.5d); + span.setTag("custom.removed", "gone"); + span.setTag("custom.removed", (Object) null); + } + + @Test + void frozenTagsSetWhatTheRecordedWritesWouldSet() { + FrozenTags frozen = FrozenTags.of((TagContributor) DDSpanFrozenTagsTest::writeTags); + DDSpan live = start(); + DDSpan replayed = start(); + try { + writeTags(live); + assertSame(replayed, replayed.setTagsFrom(frozen)); + + // compared as plain maps: TagMap.equals isn't Map content equality + assertEquals(new TreeMap<>(live.getTags()), new TreeMap<>(replayed.getTags())); + } finally { + live.finish(); + replayed.finish(); + } + } + + @Test + void everySpanGetsTheSameValueObjects() { + // the point of freezing: a value built once is shared, not rebuilt per span + TagExtractor extractor = + (n, target) -> target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0." + n); + FrozenTags frozen = FrozenTags.of(7, extractor); + DDSpan first = start(); + DDSpan second = start(); + try { + first.setTagsFrom(frozen); + second.setTagsFrom(frozen); + + assertEquals("10.0.0.7", first.getTag(Tags.PEER_HOST_IPV4)); + assertSame(first.getTag(Tags.PEER_HOST_IPV4), second.getTag(Tags.PEER_HOST_IPV4)); + } finally { + first.finish(); + second.finish(); + } + } + + @Test + void interceptedTagsKeepTheirEffects() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(DDTags.RESOURCE_NAME, "frozen-resource"); // becomes the resource + target.setTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CLIENT); // sets the cached kind + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); + DDSpan span = start(); + try { + span.setTagsFrom(frozen); + + assertEquals("frozen-resource", span.getResourceName().toString()); + assertEquals(Tags.SPAN_KIND_CLIENT, span.getSpanKindString()); + assertEquals("10.0.0.1", span.getTag(Tags.PEER_HOST_IPV4)); + } finally { + span.finish(); + } + } + + @Test + void spanKindAloneIsAppliedWithItsOrdinal() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CONSUMER); + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); + DDSpan span = start(); + try { + span.setTagsFrom(frozen); + + assertEquals(Tags.SPAN_KIND_CONSUMER, span.getTag(Tags.SPAN_KIND)); + assertEquals(Tags.SPAN_KIND_CONSUMER, span.getSpanKindString()); + } finally { + span.finish(); + } + } + + @Test + void frozenTagsOverwriteWhatTheSpanAlreadyHas() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + target.setTag(Tags.PEER_PORT, 5432); + target.setTag("custom.string", "frozen"); + }); + DDSpan span = start(); + try { + span.setTag(Tags.PEER_HOST_IPV4, "192.168.0.1"); + span.setTag("custom.string", "explicit"); + span.setTag(Tags.COMPONENT, "untouched"); + span.setTagsFrom(frozen); + + assertEquals("10.0.0.1", span.getTag(Tags.PEER_HOST_IPV4)); + assertEquals(5432, span.getTag(Tags.PEER_PORT)); + assertEquals("frozen", span.getTag("custom.string")); + assertEquals("untouched", span.getTag(Tags.COMPONENT)); + } finally { + span.finish(); + } + } + + @Test + void frozenTagsApplyAcrossTracers() { + // the core's apply plan is cached per interceptor; a second tracer must not reuse the first's + FrozenTags frozen = FrozenTags.of((TagContributor) DDSpanFrozenTagsTest::writeTags); + CoreTracer other = tracerBuilder().writer(new ListWriter()).build(); + try { + DDSpan first = start(); + DDSpan second = (DDSpan) other.startSpan("test-instr", "frozen-tags.op"); + try { + first.setTagsFrom(frozen); + second.setTagsFrom(frozen); + + assertEquals(new TreeMap<>(first.getTags()), new TreeMap<>(second.getTags())); + } finally { + first.finish(); + second.finish(); + } + } finally { + other.close(); + } + } + + @Test + void emptyOrNullFrozenTagsChangeNothing() { + DDSpan span = start(); + try { + int before = span.getTags().size(); + assertSame(span, span.setTagsFrom(FrozenTags.EMPTY)); + assertSame(span, span.setTagsFrom((FrozenTags) null)); + assertEquals(before, span.getTags().size()); + } finally { + span.finish(); + } + } +} diff --git a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java index 388d311c7b2..66debb85c4f 100644 --- a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java @@ -178,6 +178,18 @@ default AgentSpan setTagsFrom(final TagContributor contributor) { return this; } + /** + * Sets tags recorded earlier from a source that outlives the span -- the cached form of {@link + * #setTagsFrom(Object, TagExtractor)}: equivalent to replaying the recorded writes, without + * reading the source again. See {@link FrozenTags}. + */ + default AgentSpan setTagsFrom(final FrozenTags frozen) { + if (frozen != null) { + frozen.tags().forEach(this, (span, tagEntry) -> span.setTag(tagEntry)); + } + return this; + } + @Override AgentSpan setTag(String key, Number value); diff --git a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java new file mode 100644 index 00000000000..f2c1d2f8301 --- /dev/null +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java @@ -0,0 +1,207 @@ +package datadog.trace.bootstrap.instrumentation.api; + +import datadog.apicontract.Restricted; +import datadog.trace.api.TagMap; + +/** + * The tags a {@link TagExtractor} or {@link TagContributor} writes, recorded once and frozen, so a + * span can take them with {@link AgentSpan#setTagsFrom(FrozenTags)} instead of rerunning the + * extraction: the cached, cheaper form of {@code setTagsFrom}. + * + *

Use it when the source outlives the spans that take its tags and the tags are a function of it + * alone -- a remote address, a connection, a client -- and especially when reading the source + * builds new objects (an IP string, a joined list). Build it once per source with {@link + * #of(Object, TagExtractor)} and keep it wherever the source's other per-object state lives; it is + * immutable, so any thread can apply it. + * + *

Applying it is equivalent to the recorded writes, interception included. A tracer core may + * copy the tags in at once when it can show none of them needs interception or is already present. + * + *

Tag writes only: an extraction that reads the span ({@code getTag}) or sets span-level state + * ({@code setResourceName}, {@code setError}, {@code setMeasured}) can't be frozen and throws. A + * null value is recorded as no write, so unlike a live {@code setTag(tag, null)} it doesn't remove + * a tag the span already has. + */ +public final class FrozenTags { + /** No tags. */ + public static final FrozenTags EMPTY = new FrozenTags(TagMap.EMPTY); + + /** Records the tags {@code extractor} writes for {@code source}. */ + public static FrozenTags of(final T source, final TagExtractor extractor) { + if (source == null) { + return EMPTY; + } + Recorder recorder = new Recorder(); + extractor.extract(source, recorder); + return recorder.freeze(); + } + + /** Records the tags {@code contributor} writes. */ + public static FrozenTags of(final TagContributor contributor) { + if (contributor == null) { + return EMPTY; + } + Recorder recorder = new Recorder(); + contributor.addTo(recorder); + return recorder.freeze(); + } + + private final TagMap tags; // frozen + + /** + * The tracer core's plan for applying these tags, built on first use. Racy by design, like {@link + * SpanPrototype#seedPlan()}: the plan is immutable, so a reader sees null or a complete plan. + */ + private Object applyPlan; + + private FrozenTags(final TagMap tags) { + this.tags = tags; + } + + /** The recorded tags, frozen. */ + public TagMap tags() { + return tags; + } + + public boolean isEmpty() { + return tags.isEmpty(); + } + + /** The tracer core's apply plan for these tags, or null before the first span uses them. */ + @Restricted(allowedIn = {"datadog.trace.core"}) + public Object applyPlan() { + return applyPlan; + } + + /** Records the tracer core's apply plan; see {@link #applyPlan()}. */ + @Restricted(allowedIn = {"datadog.trace.core"}) + public void applyPlan(final Object plan) { + this.applyPlan = plan; + } + + /** The {@link SpanData} an extraction writes to while it is recorded. */ + private static final class Recorder implements SpanData { + private final TagMap tags = TagMap.create(); + + FrozenTags freeze() { + return tags.isEmpty() ? EMPTY : new FrozenTags(tags.freeze()); + } + + @Override + public SpanData setTag(String key, Object value) { + tags.set(key, value); + return this; + } + + @Override + public SpanData setTag(String key, String value) { + tags.set(key, (CharSequence) value); + return this; + } + + @Override + public SpanData setTag(String key, boolean value) { + tags.set(key, value); + return this; + } + + @Override + public SpanData setTag(String key, int value) { + tags.set(key, value); + return this; + } + + @Override + public SpanData setTag(String key, long value) { + tags.set(key, value); + return this; + } + + @Override + public SpanData setTag(String key, float value) { + tags.set(key, value); + return this; + } + + @Override + public SpanData setTag(String key, double value) { + tags.set(key, value); + return this; + } + + @Override + public SpanData setTag(long tagId, Object value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(long tagId, CharSequence value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(long tagId, boolean value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(long tagId, int value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(long tagId, long value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(long tagId, float value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(long tagId, double value) { + tags.set(tagId, value); + return this; + } + + @Override + public SpanData setTag(TagMap.EntryReader entry) { + tags.set(entry); + return this; + } + + @Override + public Object getTag(String key) { + // the result would depend on the span the tags are later applied to + throw new UnsupportedOperationException( + "FrozenTags records writes only; an extraction that reads the span can't be frozen"); + } + + @Override + public SpanData setResourceName(CharSequence resourceName) { + throw spanLevelState("setResourceName"); + } + + @Override + public SpanData setError(boolean error) { + throw spanLevelState("setError"); + } + + @Override + public SpanData setMeasured(boolean measured) { + throw spanLevelState("setMeasured"); + } + + private static UnsupportedOperationException spanLevelState(String method) { + return new UnsupportedOperationException( + "FrozenTags records tags only; an extraction that calls " + method + " can't be frozen"); + } + } +} diff --git a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/ImmutableSpan.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/ImmutableSpan.java index 5fcdbb6baec..3cea65e9499 100644 --- a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/ImmutableSpan.java +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/ImmutableSpan.java @@ -77,6 +77,12 @@ public AgentSpan setTagsFrom(TagContributor contributor) { return this; } + /** Skips the replay: its tags would be discarded. */ + @Override + public AgentSpan setTagsFrom(FrozenTags frozen) { + return this; + } + @Override public AgentSpan setTag(String key, Number value) { return this; diff --git a/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java b/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java new file mode 100644 index 00000000000..cb789ad491e --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java @@ -0,0 +1,107 @@ +package datadog.trace.bootstrap.instrumentation.api; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import datadog.trace.api.KnownTags; +import datadog.trace.api.TagMap; +import org.junit.jupiter.api.Test; + +class FrozenTagsTest { + + @Test + void recordsTheWritesFrozen() { + TagExtractor extractor = + (ip, span) -> { + span.setTag(KnownTags.PEER_IPV4_ID, ip); + span.setTag(Tags.PEER_PORT, 5432); + span.setTag("custom.tag", true); + }; + FrozenTags frozen = FrozenTags.of("10.0.0.1", extractor); + + assertTrue(frozen.tags().isFrozen()); + assertEquals(3, frozen.tags().size()); + assertEquals("10.0.0.1", frozen.tags().getObject(Tags.PEER_HOST_IPV4)); + assertEquals(5432, frozen.tags().getObject(Tags.PEER_PORT)); + assertEquals(true, frozen.tags().getObject("custom.tag")); + } + + @Test + void aLaterWriteOfTheSameTagWins() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + span -> { + span.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + span.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.2"); + }); + + assertEquals(1, frozen.tags().size()); + assertEquals("10.0.0.2", frozen.tags().getObject(Tags.PEER_HOST_IPV4)); + } + + @Test + void aNullValueIsNoWrite() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + span -> { + span.setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) null); + span.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); + + assertEquals(1, frozen.tags().size()); + assertNull(frozen.tags().getObject(Tags.PEER_HOSTNAME)); + } + + @Test + void nothingRecordedIsEmpty() { + assertSame(FrozenTags.EMPTY, FrozenTags.of((TagContributor) span -> {})); + assertSame(FrozenTags.EMPTY, FrozenTags.of(null, (source, span) -> span.setTag("x", 1))); + assertSame(FrozenTags.EMPTY, FrozenTags.of((TagContributor) null)); + assertTrue(FrozenTags.EMPTY.isEmpty()); + } + + @Test + void readingTheSpanCannotBeFrozen() { + assertThrows( + UnsupportedOperationException.class, + () -> FrozenTags.of((TagContributor) span -> span.getTag(Tags.PEER_HOSTNAME))); + } + + @Test + void spanLevelStateCannotBeFrozen() { + assertThrows( + UnsupportedOperationException.class, + () -> FrozenTags.of((TagContributor) span -> span.setResourceName("r"))); + assertThrows( + UnsupportedOperationException.class, + () -> FrozenTags.of((TagContributor) span -> span.setError(true))); + assertThrows( + UnsupportedOperationException.class, + () -> FrozenTags.of((TagContributor) span -> span.setMeasured(true))); + } + + @Test + void theDefaultSetTagsFromReplaysTheTags() { + FrozenTags frozen = + FrozenTags.of("10.0.0.1", (ip, span) -> span.setTag(KnownTags.PEER_IPV4_ID, ip)); + AgentSpan span = mock(AgentSpan.class, CALLS_REAL_METHODS); + + span.setTagsFrom(frozen); + + verify(span) + .setTag( + argThat( + (TagMap.EntryReader entry) -> + entry.tagId() == KnownTags.PEER_IPV4_ID + && "10.0.0.1".equals(entry.objectValue()))); + } +} From 7f0c4f7537ed7547b535868c1e4fa2d9da2b283c Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 30 Oct 2026 15:08:42 -0400 Subject: [PATCH 3/6] Freeze netty 4.1 client peer tags once per connection The client handler rebuilt peer.hostname / peer.ipv4 / peer.port from the channel's remote address on every request, allocating a new IP string each time. The address is fixed for the connection's life, so freeze the tags on first use, keep them on the connection channel (HTTP/2 streams share their parent's), and apply them with setTagsFrom(FrozenTags). Co-Authored-By: Claude Opus 5.5 --- .../HttpClientRequestTracingHandler.java | 30 ++++++++++++++++--- .../netty41/AttributeKeys.java | 5 ++++ 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java index 47134f4a0cf..ea7dec2407f 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java @@ -6,6 +6,7 @@ import static datadog.trace.instrumentation.netty41.AttributeKeys.CLIENT_PARENT_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; +import static datadog.trace.instrumentation.netty41.AttributeKeys.PEER_TAGS_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator.DECORATE; import static datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator.DECORATE_SECURE; import static datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator.NETTY_CLIENT; @@ -17,12 +18,15 @@ import datadog.context.ContextScope; import datadog.trace.api.Config; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; +import datadog.trace.bootstrap.instrumentation.decorator.PeerSocketAddress; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; import io.netty.handler.codec.http.HttpRequest; +import io.netty.util.Attribute; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -84,10 +88,7 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann decorate.afterStart(span); decorate.onRequest(span, request); - SocketAddress socketAddress = ctx.channel().remoteAddress(); - if (socketAddress instanceof InetSocketAddress) { - decorate.onPeerConnection(span, (InetSocketAddress) socketAddress); - } + span.setTagsFrom(peerTags(ctx.channel())); // AWS calls are often signed, so we can't add headers without breaking the signature. if (!awsClientCall) { @@ -111,6 +112,27 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann } } + /** + * The peer tags of the connection {@code channel} belongs to, frozen on first use and kept on the + * connection: its remote address never changes, so later requests skip rebuilding the IP string + * and resolving the hostname. HTTP/2 streams share their parent connection's tags. Not cached + * while the connection has no remote address yet. + */ + private static FrozenTags peerTags(final Channel channel) { + final Channel connection = channel.parent() != null ? channel.parent() : channel; + final Attribute attribute = connection.attr(PEER_TAGS_ATTRIBUTE_KEY); + FrozenTags tags = attribute.get(); + if (tags == null) { + final SocketAddress address = connection.remoteAddress(); + if (!(address instanceof InetSocketAddress)) { + return FrozenTags.EMPTY; + } + tags = FrozenTags.of((InetSocketAddress) address, PeerSocketAddress.INSTANCE); + attribute.set(tags); // racing requests freeze equal tags; either may win + } + return tags; + } + private static ContextContinuation takeConnectParentContinuation( final ChannelHandlerContext ctx) { final Channel channel = ctx.channel(); diff --git a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java index addf40b55d6..32d4653d8cb 100644 --- a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java +++ b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java @@ -6,6 +6,7 @@ import datadog.context.ContextContinuation; import datadog.trace.api.GenericClassValue; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext; import io.netty.util.AttributeKey; import java.util.concurrent.ConcurrentHashMap; @@ -28,6 +29,10 @@ public final class AttributeKeys { public static final AttributeKey HTTP2_CONNECTION_CODEC_ATTRIBUTE_KEY = attributeKey("datadog.http2.connection.codec"); + /** A client connection's peer tags, frozen on first use: its remote address never changes. */ + public static final AttributeKey PEER_TAGS_ATTRIBUTE_KEY = + attributeKey("datadog.client.peer.tags"); + public static final AttributeKey PARENT_CONTEXT_ATTRIBUTE_KEY = attributeKey("datadog.server.parent-context"); From cf67e596b03b0eaa49518a97fa7b06c9fe769b72 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 30 Oct 2026 15:26:52 -0400 Subject: [PATCH 4/6] Copy the non-overlapping frozen tags in bulk, overwrite the rest Frozen tags often overlap tags the span already has: an HTTP client sets peer.hostname and peer.port from the URL, then the connection's peer tags overwrite them. Falling back to tag-by-tag for the whole set made that common shape lose the bulk copy. Add a partial IdStore.seedAbsent that appends the source tags whose slot bits are clear and hands the ones that may be present to the caller, which sets them through the interceptor as setTag does. The result is the same as the recorded writes; only the overlapping tags pay for it. Co-Authored-By: Claude Opus 5.5 --- .../datadog/trace/core/DDSpanContext.java | 18 ++- .../core/DDSpanFrozenTagsDenseForkedTest.java | 43 ++++++- .../java/datadog/trace/api/DenseIdStore.java | 55 +++++++++ .../main/java/datadog/trace/api/IdStore.java | 13 +++ .../java/datadog/trace/api/TagStoresTest.java | 106 ++++++++++++++++++ 5 files changed, 226 insertions(+), 9 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java b/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java index a73496a77e1..8b9dca66a41 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java @@ -1993,16 +1993,24 @@ void setTagsFrom(final TagContributor contributor) { } /** - * Applies frozen tags as if their recorded writes were made now, under one lock: in bulk when the - * plan allows it and the masks prove none of the known tags is already on the span ({@link - * #seedKnownInBulk}), otherwise one at a time through the interceptor, as {@code setTag} does. - * Custom tags always go one at a time. Returns whether the known tags went in bulk, for tests. + * Applies frozen tags as if their recorded writes were made now, under one lock. When the plan + * allows bulk (no known tag needs the interceptor beyond span.kind), the known tags the masks + * prove absent are copied in at once and the ones that may be present -- typically set earlier + * from another source, which these overwrite -- go one at a time through the interceptor, as + * {@code setTag} does (the partial {@code IdStore.seedAbsent}); span.kind's ordinal is then set + * directly. Otherwise every tag goes one at a time. Custom tags always do. Returns whether the + * bulk path was taken, for tests. */ boolean setTagsFrom(final FrozenTags frozen) { final SeedPlan plan = seedPlanFor(frozen); final TagMap tags = frozen.tags(); synchronized (knownTags) { - if (seedKnownInBulk(plan, tags)) { + if (plan.bulk + && knownTags.seedAbsent( + tags.knownStore(), this, (ctx, tagEntry) -> ctx.putIntercepted(tagEntry))) { + if (plan.kindOrdinal != SPAN_KIND_UNSET) { + spanKindOrdinal = plan.kindOrdinal; + } HashStore custom = tags.customStore(); if (custom != null) { custom.forEach(this, (ctx, tagEntry) -> ctx.putIntercepted(tagEntry)); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java index 49c141984b7..432a47e7e37 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java @@ -12,6 +12,7 @@ import datadog.trace.bootstrap.instrumentation.api.TagContributor; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.test.junit.utils.config.WithConfig; +import java.util.TreeMap; import org.junit.jupiter.api.Test; /** @@ -90,12 +91,46 @@ void anInterceptedTagSendsTheSetOneAtATime() { } @Test - void aTagTheSpanAlreadyHasSendsTheSetOneAtATime() { + void tagsTheSpanAlreadyHasAreOverwrittenAndTheRestGoInBulk() { + // the shape of an HTTP client: peer.hostname and peer.port first from the URL, then all the + // peer tags from the connection, which win + DDSpan live = start(); + DDSpan replayed = start(); + try { + for (DDSpan span : new DDSpan[] {live, replayed}) { + span.setTag(Tags.PEER_HOSTNAME, "url.example"); + span.setTag(Tags.PEER_PORT, 80); + } + live.setTag(KnownTags.PEER_HOSTNAME_ID, "db.example"); + live.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + live.setTag(KnownTags.PEER_PORT_ID, 5432); + assertTrue(replayed.spanContext().setTagsFrom(PEER_TAGS)); + + assertEquals("db.example", replayed.getTag(Tags.PEER_HOSTNAME)); + assertEquals("10.0.0.1", replayed.getTag(Tags.PEER_HOST_IPV4)); + assertEquals(5432, replayed.getTag(Tags.PEER_PORT)); + assertEquals(new TreeMap<>(live.getTags()), new TreeMap<>(replayed.getTags())); + } finally { + live.finish(); + replayed.finish(); + } + } + + @Test + void aSpanKindTheSpanAlreadyHasIsOverwrittenWithItsOrdinal() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CLIENT); + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); DDSpan span = start(); try { - span.setTag(Tags.PEER_PORT, 1); - assertFalse(span.spanContext().setTagsFrom(PEER_TAGS)); - assertEquals(5432, span.getTag(Tags.PEER_PORT)); // still overwritten + span.setTag(Tags.SPAN_KIND, Tags.SPAN_KIND_SERVER); + assertTrue(span.spanContext().setTagsFrom(frozen)); + assertEquals(Tags.SPAN_KIND_CLIENT, span.getTag(Tags.SPAN_KIND)); + assertEquals(Tags.SPAN_KIND_CLIENT, span.getSpanKindString()); } finally { span.finish(); } diff --git a/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java b/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java index 352c64e572a..3b7ea7d5bd3 100644 --- a/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java +++ b/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java @@ -359,6 +359,61 @@ public boolean seedAbsent(IdStore source) { return true; } + /** + * The partial bulk seed: when the occupancy words share no bit, the two array copies of {@link + * #seedAbsent(IdStore)}; otherwise a walk over {@code source} that appends each tag whose slot + * bit is clear here and hands the others, which may be present, to {@code maybePresent}. The bits + * are checked against this store as it was before the walk: {@code source}'s tags are distinct, + * so one appended here can't make a later one present. + */ + @Override + public boolean seedAbsent( + IdStore source, T thisObj, BiConsumer maybePresent) { + if (this.seedAbsent(source)) { + return true; + } + if (!(source instanceof DenseIdStore)) { + return false; + } + DenseIdStore src = (DenseIdStore) source; + int n = src.used; + if (src.parent != null || n != src.count) { + return false; // read-through or tombstones: not a plain list of tags + } + long maybeHere = this.tombstoneMask | this.inheritedMask; + long spanHere = this.spanOccupancy | maybeHere; + long traceHere = this.traceOccupancy | maybeHere; + if (this.used + n > this.serials.length) { + this.growTo(this.used + n); + } + DenseReader collided = null; + for (int i = 0; i < n; i++) { + char serial = src.serials[i]; + long tagId = KnownTagCodec.idOf(serial); + long bit = slotBit(tagId); + int level = levelOf(tagId); + if (((level == TRACE_LEVEL ? traceHere : spanHere) & bit) != 0L) { + if (collided == null) { + collided = new DenseReader(); // not this.reader: maybePresent's writes reuse that + } + collided.set(tagId, src.values[i]); + maybePresent.accept(thisObj, collided); + } else { + // re-read used and the arrays: maybePresent may have appended or grown them + int at = this.used; + if (at == this.serials.length) { + this.growTo(at + 1); + } + this.serials[at] = serial; + this.values[at] = src.values[i]; + this.used = at + 1; + this.count++; + this.markLocal(bit, level); + } + } + return true; + } + /** Grows the arrays to hold at least {@code minCapacity} slots. Rare, so kept out of line. */ private void growTo(int minCapacity) { int newCap = diff --git a/internal-api/src/main/java/datadog/trace/api/IdStore.java b/internal-api/src/main/java/datadog/trace/api/IdStore.java index 74736778ce5..a4805b36a72 100644 --- a/internal-api/src/main/java/datadog/trace/api/IdStore.java +++ b/internal-api/src/main/java/datadog/trace/api/IdStore.java @@ -62,6 +62,19 @@ public boolean seedAbsent(IdStore source) { return false; } + /** + * The partial form of {@link #seedAbsent(IdStore)}: copies in the tags of {@code source} that + * can't already be present here, hands each of the rest to {@code maybePresent} for the caller to + * set tag by tag, and returns true. Returns false, having done nothing, when this store can't + * seed in bulk at all or {@code source} isn't a plain frozen list of tags; the caller then sets + * every tag itself. {@code source} must be frozen, and {@code maybePresent} must not keep the + * entry it is handed. + */ + public boolean seedAbsent( + IdStore source, T thisObj, BiConsumer maybePresent) { + return false; + } + /** * Whether {@link #seedAbsent} can ever succeed for this kind of store, so a caller can skip it, * and the lock it needs, when it can't. diff --git a/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java b/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java index 038d414b5e2..dc0f2242c95 100644 --- a/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java +++ b/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java @@ -248,6 +248,112 @@ void theHashStoreNeverSeedsInBulk() { HashIdStore target = new HashIdStore(null); assertFalse(target.seedAbsent(denseSource(KnownTags.COMPONENT_ID))); assertFalse(target.seedAbsent(new HashIdStore(null))); + assertFalse(target.seedAbsent(denseSource(KnownTags.COMPONENT_ID), target, TagStoresTest::put)); + } + + /** The caller's tag-by-tag fallback for a partial seed: an overwriting put. */ + private static void put(IdStore target, TagMap.EntryReader entry) { + target.store(entry.tagId(), entry.objectValue()); + } + + @Test + void partialSeedAppendsTheAbsentAndHandsBackTheRest() { + DenseIdStore source = + denseSource(KnownTags.COMPONENT_ID, KnownTags.SPAN_KIND_ID, KnownTags.HTTP_ROUTE_ID); + DenseIdStore target = new DenseIdStore(null); + target.store(KnownTags.COMPONENT_ID, "explicit"); + long componentBit = DenseIdStore.slotBit(KnownTags.COMPONENT_ID); + assertNotEquals(componentBit, DenseIdStore.slotBit(KnownTags.SPAN_KIND_ID), "precondition"); + assertNotEquals(componentBit, DenseIdStore.slotBit(KnownTags.HTTP_ROUTE_ID), "precondition"); + + Map handedBack = new HashMap<>(); + assertTrue( + target.seedAbsent( + source, + target, + (store, entry) -> { + handedBack.put(entry.tagId(), entry.objectValue()); + put(store, entry); + })); + + assertEquals(1, handedBack.size()); + assertEquals("component", handedBack.get(KnownTags.COMPONENT_ID)); + assertEquals(3, target.size()); + assertEquals("component", target.load(KnownTags.COMPONENT_ID)); // overwritten by the caller + assertEquals("span.kind", target.load(KnownTags.SPAN_KIND_ID)); + assertEquals("http.route", target.load(KnownTags.HTTP_ROUTE_ID)); + target.checkIntegrity(); + } + + @Test + void partialSeedCopiesInOneGoWhenNothingMayBePresent() { + DenseIdStore source = denseSource(KnownTags.COMPONENT_ID, KnownTags.SPAN_KIND_ID); + DenseIdStore target = new DenseIdStore(null); + + assertTrue( + target.seedAbsent( + source, + target, + (store, entry) -> { + throw new AssertionError("nothing may be present: " + entry.tag()); + })); + assertEquals(2, target.size()); + target.checkIntegrity(); + } + + @Test + void partialSeedHandsBackATagThatOnlySharesASlot() { + // a set bit only means "maybe present": a different tag in the same slot still goes the slow + // way, and the caller's put then appends it + long[] pair = twoSpanLevelTagsInOneSlot(); + DenseIdStore source = denseSource(pair[0]); + DenseIdStore target = new DenseIdStore(null); + target.store(pair[1], "other"); + + Map handedBack = new HashMap<>(); + assertTrue( + target.seedAbsent( + source, + target, + (store, entry) -> { + handedBack.put(entry.tagId(), entry.objectValue()); + put(store, entry); + })); + + assertEquals(1, handedBack.size()); + assertEquals(2, target.size()); + assertEquals("other", target.load(pair[1])); + assertNotNull(target.load(pair[0])); + target.checkIntegrity(); + } + + /** Two known span-level tags that share an occupancy slot. */ + private static long[] twoSpanLevelTagsInOneSlot() { + Map bySlot = new HashMap<>(); + for (int serial = 1; serial < KnownTagCodec.serialLimit(); serial++) { + long tagId = KnownTagCodec.idOf((char) serial); + if (tagId == 0 || DenseIdStore.levelOf(tagId) != DenseIdStore.SPAN_LEVEL) { + continue; + } + Long earlier = bySlot.putIfAbsent(DenseIdStore.slotBit(tagId), tagId); + if (earlier != null) { + return new long[] {tagId, earlier}; + } + } + assumeTrue(false, "no two span-level tags share a slot"); + return null; + } + + @Test + void partialSeedRefusesASourceWithAParentOrTombstones() { + DenseIdStore parent = denseSource(KnownTags.COMPONENT_ID); + DenseIdStore child = new DenseIdStore(parent); + child.store(KnownTags.SPAN_KIND_ID, "server"); + DenseIdStore target = new DenseIdStore(null); + target.store(KnownTags.SPAN_KIND_ID, "client"); // so the all-at-once copy can't take it + + assertFalse(target.seedAbsent(child, target, TagStoresTest::put), "read-through source"); + assertEquals(1, target.size(), "nothing copied"); } @Test From 5d1f8788a8866da1168c1a60cda9a09044a9b708 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 30 Oct 2026 15:40:03 -0400 Subject: [PATCH 5/6] Freeze Kafka consumer tags once per topic per poll Each consumed record's span got messaging.destination.name, kafka.group, messaging.kafka.cluster.id and messaging.kafka.bootstrap.servers set one by one, though all four are the same for every record of a topic in a poll. The poll's iterator now freezes them per topic (records come out partition by partition, so one cached topic suffices) and applies them with setTagsFrom(FrozenTags), at the same point in onConsume as before. Co-Authored-By: Claude Opus 5.5 --- .../KafkaConsumerInstrumentation.java | 1 + .../kafka_clients/KafkaConsumerTags.java | 56 ++++++++++++++++ .../kafka_clients/KafkaDecorator.java | 30 ++++----- .../kafka_clients/TracingIterator.java | 8 ++- .../kafka_clients/KafkaConsumerTagsTest.java | 66 +++++++++++++++++++ .../KafkaConsumerInstrumentation.java | 1 + .../kafka_clients38/KafkaConsumerTags.java | 56 ++++++++++++++++ .../kafka_clients38/KafkaDecorator.java | 30 ++++----- .../kafka_clients38/TracingIterator.java | 8 ++- 9 files changed, 216 insertions(+), 40 deletions(-) create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTags.java create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTagsTest.java create mode 100644 dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerTags.java diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java index 756f59aad4b..b4f4b2608a3 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java @@ -66,6 +66,7 @@ public String[] helperClassNames() { packageName + ".TextMapInjectAdapterInterface", packageName + ".KafkaConsumerInfo", packageName + ".KafkaConsumerInstrumentationHelper", + packageName + ".KafkaConsumerTags", packageName + ".KafkaDecorator", packageName + ".TextMapExtractAdapter", packageName + ".TracingIterableDelegator", diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTags.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTags.java new file mode 100644 index 00000000000..3f885fcd1e1 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTags.java @@ -0,0 +1,56 @@ +package datadog.trace.instrumentation.kafka_clients; + +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME; + +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; +import datadog.trace.bootstrap.instrumentation.api.SpanData; +import datadog.trace.bootstrap.instrumentation.api.TagExtractor; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +/** + * The consumer tags every record of one topic in one poll shares -- destination, group, cluster id + * and bootstrap servers -- extracted from the topic and frozen once, then applied to each record's + * span. Records come out of a poll partition by partition, so the topic rarely changes between + * records and one cached topic suffices. Used by one iterator, on one thread. + */ +public final class KafkaConsumerTags implements TagExtractor { + private final String group; + private final String clusterId; + private final String bootstrapServers; + + private String frozenTopic; + private FrozenTags frozen; + + public KafkaConsumerTags(String group, String clusterId, String bootstrapServers) { + this.group = group; + this.clusterId = clusterId; + this.bootstrapServers = bootstrapServers; + } + + /** The frozen tags for {@code record}'s topic, frozen when the topic changes. */ + public FrozenTags of(final ConsumerRecord record) { + final String topic = KafkaDecorator.topicOf(record); + if (topic != frozenTopic && !topic.equals(frozenTopic)) { + frozen = FrozenTags.of(topic, this); + frozenTopic = topic; + } + return frozen; + } + + @Override + public void extract(final String topic, final SpanData span) { + span.setTag(MESSAGING_DESTINATION_NAME, topic); + if (group != null) { + span.setTag(CONSUMER_GROUP, group); + } + if (clusterId != null) { + span.setTag(KAFKA_CLUSTER_ID, clusterId); + } + if (bootstrapServers != null) { + span.setTag(KAFKA_BOOTSTRAP_SERVERS, bootstrapServers); + } + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaDecorator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaDecorator.java index 53c579a4dee..5d5023eb778 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaDecorator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaDecorator.java @@ -1,6 +1,5 @@ package datadog.trace.instrumentation.kafka_clients; -import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP; import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS; import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID; import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME; @@ -16,6 +15,7 @@ import datadog.trace.api.cache.DDCaches; import datadog.trace.api.naming.SpanNaming; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; import datadog.trace.bootstrap.instrumentation.api.InternalSpanTypes; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; @@ -114,27 +114,23 @@ protected String spanKind() { return spanKind; } + /** {@code record}'s topic, or {@code "kafka"} when it has none. */ + static String topicOf(final ConsumerRecord record) { + return record.topic() == null ? "kafka" : record.topic(); + } + + /** + * Decorates a consumed record's span. {@code consumerTags} are the destination, group, cluster id + * and bootstrap servers, frozen once per topic per poll ({@link KafkaConsumerTags}). + */ public void onConsume( - final AgentSpan span, - final ConsumerRecord record, - String consumerGroup, - String clusterId, - String bootstrapServers) { + final AgentSpan span, final ConsumerRecord record, final FrozenTags consumerTags) { if (record != null) { - final String topic = record.topic() == null ? "kafka" : record.topic(); + final String topic = topicOf(record); span.setResourceName(CONSUMER_RESOURCE_NAME_CACHE.computeIfAbsent(topic, CONSUMER_PREFIX)); span.setTag(PARTITION, record.partition()); span.setTag(OFFSET, record.offset()); - span.setTag(MESSAGING_DESTINATION_NAME, topic); - if (consumerGroup != null) { - span.setTag(CONSUMER_GROUP, consumerGroup); - } - if (clusterId != null) { - span.setTag(KAFKA_CLUSTER_ID, clusterId); - } - if (bootstrapServers != null) { - span.setTag(KAFKA_BOOTSTRAP_SERVERS, bootstrapServers); - } + span.setTagsFrom(consumerTags); // TODO - do we really need both? This mechanism already adds a lot of... baggage. // check to not record a duration if the message was sent from an old Kafka client if (record.timestampType() != TimestampType.NO_TIMESTAMP_TYPE) { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java index 93179e4e3f2..2339192cee6 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java @@ -46,7 +46,9 @@ public class TracingIterator implements Iterator> { private final KafkaDecorator decorator; private final String group; private final String clusterId; - private final String bootstrapServers; + + /** This poll's consumer tags, frozen per topic. */ + private final KafkaConsumerTags consumerTags; public TracingIterator( final Iterator> delegateIterator, @@ -60,7 +62,7 @@ public TracingIterator( this.decorator = decorator; this.group = group; this.clusterId = clusterId; - this.bootstrapServers = bootstrapServers; + this.consumerTags = new KafkaConsumerTags(group, clusterId, bootstrapServers); } @Override @@ -146,7 +148,7 @@ protected void startNewRecordSpan(ConsumerRecord val) { span.setTag(InstrumentationTags.TOMBSTONE, true); } decorator.afterStart(span); - decorator.onConsume(span, val, group, clusterId, bootstrapServers); + decorator.onConsume(span, val, consumerTags.of(val)); if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { activateNext(span); } else { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTagsTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTagsTest.java new file mode 100644 index 00000000000..b22a69c73dc --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/test/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerTagsTest.java @@ -0,0 +1,66 @@ +package datadog.trace.instrumentation.kafka_clients; + +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import datadog.trace.api.TagMap; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.junit.jupiter.api.Test; + +class KafkaConsumerTagsTest { + private static ConsumerRecord record(String topic, long offset) { + return new ConsumerRecord<>(topic, 0, offset, "key", "value"); + } + + @Test + void freezesTheConsumerTagsOfATopic() { + KafkaConsumerTags tags = new KafkaConsumerTags("group", "cluster", "broker:9092"); + + TagMap frozen = tags.of(record("orders", 0)).tags(); + + assertEquals(4, frozen.size()); + assertEquals("orders", frozen.getObject(MESSAGING_DESTINATION_NAME)); + assertEquals("group", frozen.getObject(CONSUMER_GROUP)); + assertEquals("cluster", frozen.getObject(KAFKA_CLUSTER_ID)); + assertEquals("broker:9092", frozen.getObject(KAFKA_BOOTSTRAP_SERVERS)); + } + + @Test + void recordsOfTheSameTopicShareOneFreeze() { + KafkaConsumerTags tags = new KafkaConsumerTags("group", "cluster", "broker:9092"); + + FrozenTags first = tags.of(record("orders", 0)); + // an equal but distinct topic string, as a later fetch may hand out + assertSame(first, tags.of(record(new String("orders"), 1))); + } + + @Test + void aNewTopicFreezesAgain() { + KafkaConsumerTags tags = new KafkaConsumerTags("group", "cluster", "broker:9092"); + + FrozenTags orders = tags.of(record("orders", 0)); + FrozenTags payments = tags.of(record("payments", 0)); + + assertNotSame(orders, payments); + assertEquals("payments", payments.tags().getObject(MESSAGING_DESTINATION_NAME)); + assertEquals( + "orders", tags.of(record("orders", 1)).tags().getObject(MESSAGING_DESTINATION_NAME)); + } + + @Test + void missingConsumerInfoIsLeftOut() { + KafkaConsumerTags tags = new KafkaConsumerTags(null, null, null); + + TagMap frozen = tags.of(record("orders", 0)).tags(); + + assertEquals(1, frozen.size()); + assertNull(frozen.getObject(CONSUMER_GROUP)); + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentation.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentation.java index 70f3f6dbc92..4da2b5be2a6 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentation.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerInstrumentation.java @@ -54,6 +54,7 @@ public String[] helperClassNames() { packageName + ".TextMapInjectAdapterInterface", packageName + ".KafkaConsumerInfo", packageName + ".KafkaConsumerInstrumentationHelper", + packageName + ".KafkaConsumerTags", packageName + ".KafkaDecorator", packageName + ".TextMapExtractAdapter", packageName + ".TracingIterableDelegator", diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerTags.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerTags.java new file mode 100644 index 00000000000..70eeecbd772 --- /dev/null +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaConsumerTags.java @@ -0,0 +1,56 @@ +package datadog.trace.instrumentation.kafka_clients38; + +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME; + +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; +import datadog.trace.bootstrap.instrumentation.api.SpanData; +import datadog.trace.bootstrap.instrumentation.api.TagExtractor; +import org.apache.kafka.clients.consumer.ConsumerRecord; + +/** + * The consumer tags every record of one topic in one poll shares -- destination, group, cluster id + * and bootstrap servers -- extracted from the topic and frozen once, then applied to each record's + * span. Records come out of a poll partition by partition, so the topic rarely changes between + * records and one cached topic suffices. Used by one iterator, on one thread. + */ +public final class KafkaConsumerTags implements TagExtractor { + private final String group; + private final String clusterId; + private final String bootstrapServers; + + private String frozenTopic; + private FrozenTags frozen; + + public KafkaConsumerTags(String group, String clusterId, String bootstrapServers) { + this.group = group; + this.clusterId = clusterId; + this.bootstrapServers = bootstrapServers; + } + + /** The frozen tags for {@code record}'s topic, frozen when the topic changes. */ + public FrozenTags of(final ConsumerRecord record) { + final String topic = KafkaDecorator.topicOf(record); + if (topic != frozenTopic && !topic.equals(frozenTopic)) { + frozen = FrozenTags.of(topic, this); + frozenTopic = topic; + } + return frozen; + } + + @Override + public void extract(final String topic, final SpanData span) { + span.setTag(MESSAGING_DESTINATION_NAME, topic); + if (group != null) { + span.setTag(CONSUMER_GROUP, group); + } + if (clusterId != null) { + span.setTag(KAFKA_CLUSTER_ID, clusterId); + } + if (bootstrapServers != null) { + span.setTag(KAFKA_BOOTSTRAP_SERVERS, bootstrapServers); + } + } +} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaDecorator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaDecorator.java index d2d6f53b8a9..d09533cd8a2 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaDecorator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/KafkaDecorator.java @@ -1,6 +1,5 @@ package datadog.trace.instrumentation.kafka_clients38; -import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.CONSUMER_GROUP; import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_BOOTSTRAP_SERVERS; import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.KAFKA_CLUSTER_ID; import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.MESSAGING_DESTINATION_NAME; @@ -16,6 +15,7 @@ import datadog.trace.api.cache.DDCaches; import datadog.trace.api.naming.SpanNaming; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.FrozenTags; import datadog.trace.bootstrap.instrumentation.api.InternalSpanTypes; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; @@ -114,27 +114,23 @@ protected String spanKind() { return spanKind; } + /** {@code record}'s topic, or {@code "kafka"} when it has none. */ + static String topicOf(final ConsumerRecord record) { + return record.topic() == null ? "kafka" : record.topic(); + } + + /** + * Decorates a consumed record's span. {@code consumerTags} are the destination, group, cluster id + * and bootstrap servers, frozen once per topic per poll ({@link KafkaConsumerTags}). + */ public void onConsume( - final AgentSpan span, - final ConsumerRecord record, - String consumerGroup, - String clusterId, - String bootstrapServers) { + final AgentSpan span, final ConsumerRecord record, final FrozenTags consumerTags) { if (record != null) { - final String topic = record.topic() == null ? "kafka" : record.topic(); + final String topic = topicOf(record); span.setResourceName(CONSUMER_RESOURCE_NAME_CACHE.computeIfAbsent(topic, CONSUMER_PREFIX)); span.setTag(PARTITION, record.partition()); span.setTag(OFFSET, record.offset()); - span.setTag(MESSAGING_DESTINATION_NAME, topic); - if (consumerGroup != null) { - span.setTag(CONSUMER_GROUP, consumerGroup); - } - if (clusterId != null) { - span.setTag(KAFKA_CLUSTER_ID, clusterId); - } - if (bootstrapServers != null) { - span.setTag(KAFKA_BOOTSTRAP_SERVERS, bootstrapServers); - } + span.setTagsFrom(consumerTags); // TODO - do we really need both? This mechanism already adds a lot of... baggage. // check to not record a duration if the message was sent from an old Kafka client if (record.timestampType() != TimestampType.NO_TIMESTAMP_TYPE) { diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java index 7beba848473..d92426ae035 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/TracingIterator.java @@ -42,7 +42,9 @@ public class TracingIterator implements Iterator> { private final KafkaDecorator decorator; private final String group; private final String clusterId; - private final String bootstrapServers; + + /** This poll's consumer tags, frozen per topic. */ + private final KafkaConsumerTags consumerTags; public TracingIterator( final Iterator> delegateIterator, @@ -56,7 +58,7 @@ public TracingIterator( this.decorator = decorator; this.group = group; this.clusterId = clusterId; - this.bootstrapServers = bootstrapServers; + this.consumerTags = new KafkaConsumerTags(group, clusterId, bootstrapServers); } @Override @@ -142,7 +144,7 @@ protected void startNewRecordSpan(ConsumerRecord val) { span.setTag(InstrumentationTags.TOMBSTONE, true); } decorator.afterStart(span); - decorator.onConsume(span, val, group, clusterId, bootstrapServers); + decorator.onConsume(span, val, consumerTags.of(val)); if (InstrumenterConfig.get().isLegacyContextManagerEnabled()) { activateNext(span); } else { From 3ddcf66479cbbc43b0b312d4e947ff471726af55 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 30 Oct 2026 17:06:59 -0400 Subject: [PATCH 6/6] Replay frozen tags in write order and keep peer DNS outside the tag lock Review fixes, keeping FrozenTags exactly equivalent to the live writes: - FrozenTags records each write (same setter, same value) and replays them in order on every tag-by-tag path. Split-by-tags is last-write- wins, so replaying in store order could pick a different service. A recorded null or empty value now replays as written instead of being dropped, and keeps the set off the bulk path, as does a custom tag that split-by-tags routes. - BaseDecorator.onPeerConnection calls the peer extractors directly: through setTagsFrom, the hostname's reverse DNS lookup ran under the span's tag lock. - netty reads the address from the request's channel as before, shares the connection's frozen tags only when the parent reports the same address object (HTTP/2 streams, not a server channel), and refreezes when the address changes. - Rename the partial seed to IdStore.seedPartial. Co-Authored-By: Claude Opus 5.5 --- .../decorator/BaseDecorator.java | 12 ++- .../decorator/PeerAddress.java | 4 + .../HttpClientRequestTracingHandler.java | 33 ++++--- .../netty41/AttributeKeys.java | 7 +- .../datadog/trace/core/DDSpanContext.java | 42 +++++++-- .../trace/core/DDSpanFrozenTagsTest.java | 74 +++++++++++++++ .../java/datadog/trace/api/DenseIdStore.java | 2 +- .../main/java/datadog/trace/api/IdStore.java | 2 +- .../instrumentation/api/AgentSpan.java | 2 +- .../instrumentation/api/FrozenTags.java | 91 ++++++++++++++++--- .../java/datadog/trace/api/TagStoresTest.java | 11 ++- .../instrumentation/api/FrozenTagsTest.java | 58 +++++++++--- 12 files changed, 281 insertions(+), 57 deletions(-) diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java index e889750638e..cccac3c697e 100644 --- a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/BaseDecorator.java @@ -190,9 +190,15 @@ protected void doOnError(final AgentSpan span, final Throwable throwable, byte e throwable instanceof ExecutionException ? throwable.getCause() : throwable, errorPriority); } + // The peer extractors write straight to the span rather than through setTagsFrom: resolving the + // hostname can mean a reverse DNS lookup, which mustn't run under the span's tag lock. To skip + // the lookup per span, freeze the extraction once per address instead (FrozenTags). + public final void onPeerConnection( final AgentSpan span, @Nullable final InetSocketAddress remoteConnection) { - span.setTagsFrom(remoteConnection, PeerSocketAddress.INSTANCE); + if (remoteConnection != null) { + PeerSocketAddress.INSTANCE.extract(remoteConnection, span); + } } public final void onPeerConnection( @@ -202,7 +208,9 @@ public final void onPeerConnection( public final void onPeerConnection( AgentSpan span, @Nullable InetAddress remoteAddress, boolean resolved) { - span.setTagsFrom(remoteAddress, resolved ? PeerAddress.RESOLVED : PeerAddress.UNRESOLVED); + if (remoteAddress != null) { + (resolved ? PeerAddress.RESOLVED : PeerAddress.UNRESOLVED).extract(remoteAddress, span); + } } public void setPeerPort(AgentSpan span, String port) { diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java index 7dac17ab4a6..35691fade5d 100644 --- a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java @@ -14,6 +14,10 @@ * Extracts the peer tags of a remote {@link InetAddress}: its IP, as {@code peer.ipv4} or {@code * peer.ipv6}, and, when the address was resolved and peer hostnames are enabled, {@code * peer.hostname}. + * + *

Resolving the hostname may do a reverse DNS lookup, so don't apply this with {@code + * span.setTagsFrom(address, extractor)}, which holds the span's tag lock while it runs: call {@code + * extract} on the span directly, or freeze it once per address ({@code FrozenTags.of}). */ public final class PeerAddress implements TagExtractor { /** For an address that was resolved from a hostname. */ diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java index ea7dec2407f..c61ed2d7d37 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/client/HttpClientRequestTracingHandler.java @@ -6,6 +6,7 @@ import static datadog.trace.instrumentation.netty41.AttributeKeys.CLIENT_PARENT_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; +import static datadog.trace.instrumentation.netty41.AttributeKeys.PEER_TAGS_ADDRESS_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.PEER_TAGS_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator.DECORATE; import static datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator.DECORATE_SECURE; @@ -113,22 +114,28 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann } /** - * The peer tags of the connection {@code channel} belongs to, frozen on first use and kept on the - * connection: its remote address never changes, so later requests skip rebuilding the IP string - * and resolving the hostname. HTTP/2 streams share their parent connection's tags. Not cached - * while the connection has no remote address yet. + * The peer tags of {@code channel}'s remote address, frozen once per address and kept on the + * channel, so later requests skip rebuilding the IP string and resolving the hostname. An HTTP/2 + * stream reports its connection's address object, so its tags are kept on the connection and + * shared by its streams; any other parent (a server channel) reports a different address, so the + * channel keeps its own. The cached tags are reused only while the channel still reports the + * address object they were frozen from. Nothing is cached while there is no remote address yet. + * Called on the channel's event loop, which HTTP/2 streams share with their connection. */ private static FrozenTags peerTags(final Channel channel) { - final Channel connection = channel.parent() != null ? channel.parent() : channel; - final Attribute attribute = connection.attr(PEER_TAGS_ATTRIBUTE_KEY); - FrozenTags tags = attribute.get(); - if (tags == null) { - final SocketAddress address = connection.remoteAddress(); - if (!(address instanceof InetSocketAddress)) { - return FrozenTags.EMPTY; - } + final SocketAddress address = channel.remoteAddress(); + if (!(address instanceof InetSocketAddress)) { + return FrozenTags.EMPTY; + } + final Channel parent = channel.parent(); + final Channel holder = parent != null && parent.remoteAddress() == address ? parent : channel; + final Attribute tagsAttribute = holder.attr(PEER_TAGS_ATTRIBUTE_KEY); + final Attribute addressAttribute = holder.attr(PEER_TAGS_ADDRESS_ATTRIBUTE_KEY); + FrozenTags tags = tagsAttribute.get(); + if (tags == null || addressAttribute.get() != address) { tags = FrozenTags.of((InetSocketAddress) address, PeerSocketAddress.INSTANCE); - attribute.set(tags); // racing requests freeze equal tags; either may win + tagsAttribute.set(tags); + addressAttribute.set(address); } return tags; } diff --git a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java index 32d4653d8cb..d70aafc2226 100644 --- a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java +++ b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java @@ -9,6 +9,7 @@ import datadog.trace.bootstrap.instrumentation.api.FrozenTags; import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext; import io.netty.util.AttributeKey; +import java.net.SocketAddress; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -29,10 +30,14 @@ public final class AttributeKeys { public static final AttributeKey HTTP2_CONNECTION_CODEC_ATTRIBUTE_KEY = attributeKey("datadog.http2.connection.codec"); - /** A client connection's peer tags, frozen on first use: its remote address never changes. */ + /** A client channel's peer tags, frozen from its remote address. */ public static final AttributeKey PEER_TAGS_ATTRIBUTE_KEY = attributeKey("datadog.client.peer.tags"); + /** The remote address {@link #PEER_TAGS_ATTRIBUTE_KEY}'s tags were frozen from. */ + public static final AttributeKey PEER_TAGS_ADDRESS_ATTRIBUTE_KEY = + attributeKey("datadog.client.peer.tags.address"); + public static final AttributeKey PARENT_CONTEXT_ATTRIBUTE_KEY = attributeKey("datadog.server.parent-context"); diff --git a/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java b/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java index 8b9dca66a41..535d0c0f2b0 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java @@ -798,11 +798,34 @@ private SeedPlan seedPlanFor(final FrozenTags frozen) { if (isCurrentPlan(cached)) { return (SeedPlan) cached; } - SeedPlan plan = newSeedPlan(frozen.tags()); + TagMap tags = frozen.tags(); + SeedPlan plan = newSeedPlan(tags); + if (plan.bulk && (frozen.isReplayOnly() || anyCustomTagIntercepted(tags))) { + // copying the net result in wouldn't be the same as the writes: replay them instead + plan = new SeedPlan(tagInterceptor, false, plan.kindOrdinal); + } frozen.applyPlan(plan); return plan; } + /** + * Whether split-by-tags routes any of {@code tags}' custom tags. canSeedInBulk leaves custom tags + * out, since a prototype seeds them one at a time; frozen tags in bulk set them in no particular + * order, which only matters when they're routed, the last routed write deciding the service. + */ + private boolean anyCustomTagIntercepted(final TagMap tags) { + HashStore custom = tags.customStore(); + if (custom == null) { + return false; + } + for (final TagMap.EntryReader tagEntry : tags) { + if (tagEntry.tagId() == 0 && tagInterceptor.needsInterceptCustomTag(tagEntry.tag())) { + return true; + } + } + return false; + } + private boolean isCurrentPlan(final Object cached) { return cached instanceof SeedPlan && ((SeedPlan) cached).interceptor == tagInterceptor; } @@ -1994,19 +2017,20 @@ void setTagsFrom(final TagContributor contributor) { /** * Applies frozen tags as if their recorded writes were made now, under one lock. When the plan - * allows bulk (no known tag needs the interceptor beyond span.kind), the known tags the masks - * prove absent are copied in at once and the ones that may be present -- typically set earlier - * from another source, which these overwrite -- go one at a time through the interceptor, as - * {@code setTag} does (the partial {@code IdStore.seedAbsent}); span.kind's ordinal is then set - * directly. Otherwise every tag goes one at a time. Custom tags always do. Returns whether the - * bulk path was taken, for tests. + * allows bulk -- no write needs the interceptor beyond span.kind, and none removes a tag or sets + * an empty value -- their net result is copied in: the known tags the masks prove absent at once, + * the ones that may be present (typically set earlier from another source, which these overwrite) + * one at a time through the interceptor, as {@code setTag} does (the partial {@code + * IdStore.seedPartial}), then span.kind's ordinal directly and the custom tags one at a time. + * Otherwise the writes are replayed in order, exactly as the extraction made them. Returns + * whether the bulk path was taken, for tests. */ boolean setTagsFrom(final FrozenTags frozen) { final SeedPlan plan = seedPlanFor(frozen); final TagMap tags = frozen.tags(); synchronized (knownTags) { if (plan.bulk - && knownTags.seedAbsent( + && knownTags.seedPartial( tags.knownStore(), this, (ctx, tagEntry) -> ctx.putIntercepted(tagEntry))) { if (plan.kindOrdinal != SPAN_KIND_UNSET) { spanKindOrdinal = plan.kindOrdinal; @@ -2017,7 +2041,7 @@ boolean setTagsFrom(final FrozenTags frozen) { } return true; } - tags.forEach(this, (ctx, tagEntry) -> ctx.putIntercepted(tagEntry)); + frozen.replay(new HeldSpanData(this)); return false; } } diff --git a/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java index 54bba7717ab..36a9e5362f9 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java @@ -1,6 +1,8 @@ package datadog.trace.core; +import static java.util.Arrays.asList; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import datadog.trace.api.DDTags; @@ -11,6 +13,10 @@ import datadog.trace.bootstrap.instrumentation.api.TagExtractor; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.common.writer.ListWriter; +import datadog.trace.core.taginterceptor.RuleFlags; +import datadog.trace.core.taginterceptor.TagInterceptor; +import java.util.HashSet; +import java.util.Set; import java.util.TreeMap; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -197,4 +203,72 @@ void emptyOrNullFrozenTagsChangeNothing() { span.finish(); } } + + @Test + void aRecordedRemovalRemovesTheSpansTag() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + target -> { + target.setTag("custom.stale", (Object) null); + target.setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) null); + target.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + }); + DDSpan span = start(); + try { + span.setTag("custom.stale", "stale"); + span.setTag(Tags.PEER_HOSTNAME, "url.example"); + span.setTagsFrom(frozen); + + assertNull(span.getTag("custom.stale")); + assertNull(span.getTag(Tags.PEER_HOSTNAME)); + assertEquals("10.0.0.1", span.getTag(Tags.PEER_HOST_IPV4)); + } finally { + span.finish(); + } + } + + @Test + void splitByTagsSeesTheWritesInTheirOrder() { + // split-by-tags makes each routed write set the service, so the last one decides it: the + // replay must make the writes in the extraction's order, not the store's + Set splitTags = new HashSet<>(asList("custom.first", "custom.second", "custom.third")); + CoreTracer splitting = + tracerBuilder() + .writer(new ListWriter()) + .tagInterceptor( + new TagInterceptor(true, "my-service", splitTags, new RuleFlags(), false)) + .build(); + try { + for (String[] order : + new String[][] { + {"custom.first", "custom.second", "custom.third"}, + {"custom.third", "custom.second", "custom.first"}, + {"custom.second", "custom.third", "custom.first"} + }) { + TagContributor writes = + target -> { + for (String tag : order) { + target.setTag(tag, tag + "-service"); + } + }; + FrozenTags frozen = FrozenTags.of(writes); + DDSpan live = (DDSpan) splitting.startSpan("test-instr", "frozen-tags.op"); + DDSpan replayed = (DDSpan) splitting.startSpan("test-instr", "frozen-tags.op"); + try { + writes.addTo(live); + replayed.setTagsFrom(frozen); + + assertEquals(order[2] + "-service", live.getServiceName()); + assertEquals(live.getServiceName(), replayed.getServiceName()); + assertEquals(new TreeMap<>(live.getTags()), new TreeMap<>(replayed.getTags())); + } finally { + live.finish(); + replayed.finish(); + } + } + } finally { + splitting.close(); + } + } } diff --git a/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java b/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java index 3b7ea7d5bd3..0a331a8d051 100644 --- a/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java +++ b/internal-api/src/main/java/datadog/trace/api/DenseIdStore.java @@ -367,7 +367,7 @@ public boolean seedAbsent(IdStore source) { * so one appended here can't make a later one present. */ @Override - public boolean seedAbsent( + public boolean seedPartial( IdStore source, T thisObj, BiConsumer maybePresent) { if (this.seedAbsent(source)) { return true; diff --git a/internal-api/src/main/java/datadog/trace/api/IdStore.java b/internal-api/src/main/java/datadog/trace/api/IdStore.java index a4805b36a72..36727a308f1 100644 --- a/internal-api/src/main/java/datadog/trace/api/IdStore.java +++ b/internal-api/src/main/java/datadog/trace/api/IdStore.java @@ -70,7 +70,7 @@ public boolean seedAbsent(IdStore source) { * every tag itself. {@code source} must be frozen, and {@code maybePresent} must not keep the * entry it is handed. */ - public boolean seedAbsent( + public boolean seedPartial( IdStore source, T thisObj, BiConsumer maybePresent) { return false; } diff --git a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java index 66debb85c4f..a9a4a8cc52f 100644 --- a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java @@ -185,7 +185,7 @@ default AgentSpan setTagsFrom(final TagContributor contributor) { */ default AgentSpan setTagsFrom(final FrozenTags frozen) { if (frozen != null) { - frozen.tags().forEach(this, (span, tagEntry) -> span.setTag(tagEntry)); + frozen.replay(this); } return this; } diff --git a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java index f2c1d2f8301..f9b865a65be 100644 --- a/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java @@ -2,6 +2,8 @@ import datadog.apicontract.Restricted; import datadog.trace.api.TagMap; +import java.util.ArrayList; +import java.util.List; /** * The tags a {@link TagExtractor} or {@link TagContributor} writes, recorded once and frozen, so a @@ -14,17 +16,17 @@ * #of(Object, TagExtractor)} and keep it wherever the source's other per-object state lives; it is * immutable, so any thread can apply it. * - *

Applying it is equivalent to the recorded writes, interception included. A tracer core may - * copy the tags in at once when it can show none of them needs interception or is already present. + *

Applying it is equivalent to making the recorded writes again, in order, interception + * included: it {@linkplain #replay replays} them, or a tracer core copies their net result in at + * once where it can show that's the same -- no write needs interception and none removes a tag (a + * null value) or sets an empty one. * *

Tag writes only: an extraction that reads the span ({@code getTag}) or sets span-level state - * ({@code setResourceName}, {@code setError}, {@code setMeasured}) can't be frozen and throws. A - * null value is recorded as no write, so unlike a live {@code setTag(tag, null)} it doesn't remove - * a tag the span already has. + * ({@code setResourceName}, {@code setError}, {@code setMeasured}) can't be frozen and throws. */ public final class FrozenTags { /** No tags. */ - public static final FrozenTags EMPTY = new FrozenTags(TagMap.EMPTY); + public static final FrozenTags EMPTY = new FrozenTags(TagMap.EMPTY, new Write[0], false); /** Records the tags {@code extractor} writes for {@code source}. */ public static FrozenTags of(final T source, final TagExtractor extractor) { @@ -46,7 +48,19 @@ public static FrozenTags of(final TagContributor contributor) { return recorder.freeze(); } - private final TagMap tags; // frozen + /** One recorded write: the same setter with the same value. */ + private interface Write { + void replay(SpanData span); + } + + /** The net result of the writes, frozen: what a bulk copy copies. */ + private final TagMap tags; + + /** The writes in the order they were made. */ + private final Write[] writes; + + /** Some write removed a tag or set an empty value, so only a replay is equivalent. */ + private final boolean replayOnly; /** * The tracer core's plan for applying these tags, built on first use. Racy by design, like {@link @@ -54,17 +68,35 @@ public static FrozenTags of(final TagContributor contributor) { */ private Object applyPlan; - private FrozenTags(final TagMap tags) { + private FrozenTags(final TagMap tags, final Write[] writes, final boolean replayOnly) { this.tags = tags; + this.writes = writes; + this.replayOnly = replayOnly; } - /** The recorded tags, frozen. */ + /** The net result of the recorded writes, frozen. */ public TagMap tags() { return tags; } + /** Whether nothing was written. */ public boolean isEmpty() { - return tags.isEmpty(); + return writes.length == 0; + } + + /** + * Whether only {@link #replay} is equivalent to the writes: some write removed a tag or set an + * empty value, which copying {@link #tags} in wouldn't do. + */ + public boolean isReplayOnly() { + return replayOnly; + } + + /** Makes the recorded writes on {@code span}, in order. */ + public void replay(final SpanData span) { + for (Write write : writes) { + write.replay(span); + } } /** The tracer core's apply plan for these tags, or null before the first span uses them. */ @@ -82,98 +114,133 @@ public void applyPlan(final Object plan) { /** The {@link SpanData} an extraction writes to while it is recorded. */ private static final class Recorder implements SpanData { private final TagMap tags = TagMap.create(); + private final List writes = new ArrayList<>(); + private boolean replayOnly; FrozenTags freeze() { - return tags.isEmpty() ? EMPTY : new FrozenTags(tags.freeze()); + return writes.isEmpty() + ? EMPTY + : new FrozenTags(tags.freeze(), writes.toArray(new Write[0]), replayOnly); + } + + /** Notes a write whose value a bulk copy of the net result wouldn't reproduce. */ + private void check(final Object value) { + if (TagMap.Entry.isEmptyValue(value)) { + replayOnly = true; // null removes the tag; an empty value is stored + } } @Override public SpanData setTag(String key, Object value) { + check(value); tags.set(key, value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(String key, String value) { + check(value); tags.set(key, (CharSequence) value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(String key, boolean value) { tags.set(key, value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(String key, int value) { tags.set(key, value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(String key, long value) { tags.set(key, value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(String key, float value) { tags.set(key, value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(String key, double value) { tags.set(key, value); + writes.add(span -> span.setTag(key, value)); return this; } @Override public SpanData setTag(long tagId, Object value) { + check(value); tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(long tagId, CharSequence value) { + check(value); tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(long tagId, boolean value) { tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(long tagId, int value) { tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(long tagId, long value) { tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(long tagId, float value) { tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(long tagId, double value) { tags.set(tagId, value); + writes.add(span -> span.setTag(tagId, value)); return this; } @Override public SpanData setTag(TagMap.EntryReader entry) { - tags.set(entry); + if (entry == null) { + return this; // no change, as on a span + } + // the reader may be a reused flyweight: keep an entry of its own + TagMap.Entry copy = entry.entry(); + tags.set(copy); + writes.add(span -> span.setTag(copy)); return this; } diff --git a/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java b/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java index dc0f2242c95..c843a427248 100644 --- a/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java +++ b/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java @@ -248,7 +248,8 @@ void theHashStoreNeverSeedsInBulk() { HashIdStore target = new HashIdStore(null); assertFalse(target.seedAbsent(denseSource(KnownTags.COMPONENT_ID))); assertFalse(target.seedAbsent(new HashIdStore(null))); - assertFalse(target.seedAbsent(denseSource(KnownTags.COMPONENT_ID), target, TagStoresTest::put)); + assertFalse( + target.seedPartial(denseSource(KnownTags.COMPONENT_ID), target, TagStoresTest::put)); } /** The caller's tag-by-tag fallback for a partial seed: an overwriting put. */ @@ -268,7 +269,7 @@ void partialSeedAppendsTheAbsentAndHandsBackTheRest() { Map handedBack = new HashMap<>(); assertTrue( - target.seedAbsent( + target.seedPartial( source, target, (store, entry) -> { @@ -291,7 +292,7 @@ void partialSeedCopiesInOneGoWhenNothingMayBePresent() { DenseIdStore target = new DenseIdStore(null); assertTrue( - target.seedAbsent( + target.seedPartial( source, target, (store, entry) -> { @@ -312,7 +313,7 @@ void partialSeedHandsBackATagThatOnlySharesASlot() { Map handedBack = new HashMap<>(); assertTrue( - target.seedAbsent( + target.seedPartial( source, target, (store, entry) -> { @@ -352,7 +353,7 @@ void partialSeedRefusesASourceWithAParentOrTombstones() { DenseIdStore target = new DenseIdStore(null); target.store(KnownTags.SPAN_KIND_ID, "client"); // so the all-at-once copy can't take it - assertFalse(target.seedAbsent(child, target, TagStoresTest::put), "read-through source"); + assertFalse(target.seedPartial(child, target, TagStoresTest::put), "read-through source"); assertEquals(1, target.size(), "nothing copied"); } diff --git a/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java b/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java index cb789ad491e..8318f587adb 100644 --- a/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java +++ b/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java @@ -1,18 +1,20 @@ package datadog.trace.bootstrap.instrumentation.api; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import datadog.trace.api.KnownTags; -import datadog.trace.api.TagMap; import org.junit.jupiter.api.Test; +import org.mockito.InOrder; class FrozenTagsTest { @@ -48,17 +50,54 @@ void aLaterWriteOfTheSameTagWins() { } @Test - void aNullValueIsNoWrite() { - FrozenTags frozen = + void aNullOrEmptyValueMakesTheTagsReplayOnly() { + // a null removes the tag and an empty value is stored, which copying the net result in + // wouldn't reproduce + FrozenTags removing = FrozenTags.of( (TagContributor) span -> { span.setTag(KnownTags.PEER_HOSTNAME_ID, (CharSequence) null); span.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); }); + assertTrue(removing.isReplayOnly()); + assertEquals(1, removing.tags().size()); + assertNull(removing.tags().getObject(Tags.PEER_HOSTNAME)); - assertEquals(1, frozen.tags().size()); - assertNull(frozen.tags().getObject(Tags.PEER_HOSTNAME)); + assertTrue(FrozenTags.of((TagContributor) span -> span.setTag("custom", "")).isReplayOnly()); + assertFalse(FrozenTags.of((TagContributor) span -> span.setTag("custom", "x")).isReplayOnly()); + } + + @Test + void aRemovalAloneIsNotEmpty() { + FrozenTags frozen = + FrozenTags.of((TagContributor) span -> span.setTag(Tags.PEER_HOSTNAME, (Object) null)); + + assertFalse(frozen.isEmpty()); + assertTrue(frozen.tags().isEmpty()); + } + + @Test + void replayMakesTheWritesInOrderWithTheirOwnSetters() { + FrozenTags frozen = + FrozenTags.of( + (TagContributor) + span -> { + span.setTag("custom.b", 2); + span.setTag(KnownTags.PEER_IPV4_ID, "10.0.0.1"); + span.setTag("custom.a", (Object) null); + span.setTag("custom.b", 3L); + }); + SpanData span = mock(SpanData.class); + + frozen.replay(span); + + InOrder order = inOrder(span); + order.verify(span).setTag("custom.b", 2); + order.verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); + order.verify(span).setTag("custom.a", (Object) null); + order.verify(span).setTag("custom.b", 3L); + verifyNoMoreInteractions(span); } @Test @@ -97,11 +136,6 @@ void theDefaultSetTagsFromReplaysTheTags() { span.setTagsFrom(frozen); - verify(span) - .setTag( - argThat( - (TagMap.EntryReader entry) -> - entry.tagId() == KnownTags.PEER_IPV4_ID - && "10.0.0.1".equals(entry.objectValue()))); + verify(span).setTag(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); } }