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..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 @@ -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; @@ -194,11 +190,14 @@ 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) { if (remoteConnection != null) { - onPeerConnection(span, remoteConnection.getAddress(), !remoteConnection.isUnresolved()); - setPeerPort(span, remoteConnection.getPort()); + PeerSocketAddress.INSTANCE.extract(remoteConnection, span); } } @@ -210,15 +209,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); - } + (resolved ? PeerAddress.RESOLVED : PeerAddress.UNRESOLVED).extract(remoteAddress, span); } } 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..35691fade5d --- /dev/null +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/PeerAddress.java @@ -0,0 +1,47 @@ +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}. + * + *

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. */ + 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()); + } +} 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 { 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..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,8 @@ 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; import static datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator.NETTY_CLIENT; @@ -17,12 +19,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 +89,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 +113,33 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann } } + /** + * 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 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); + tagsAttribute.set(tags); + addressAttribute.set(address); + } + 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..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 @@ -6,8 +6,10 @@ 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.net.SocketAddress; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -28,6 +30,14 @@ public final class AttributeKeys { public static final AttributeKey HTTP2_CONNECTION_CODEC_ATTRIBUTE_KEY = attributeKey("datadog.http2.connection.codec"); + /** 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/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..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 @@ -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,59 @@ 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; + } + 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; + } + + 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 +2015,37 @@ 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 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.seedPartial( + 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)); + } + return true; + } + frozen.replay(new HeldSpanData(this)); + 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..432a47e7e37 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsDenseForkedTest.java @@ -0,0 +1,138 @@ +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 java.util.TreeMap; +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 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.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/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..36a9e5362f9 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/DDSpanFrozenTagsTest.java @@ -0,0 +1,274 @@ +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; +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 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; +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(); + } + } + + @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 352c64e572a..0a331a8d051 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 seedPartial( + 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..36727a308f1 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 seedPartial( + 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/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/AgentSpan.java index 388d311c7b2..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 @@ -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.replay(this); + } + 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..f9b865a65be --- /dev/null +++ b/internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/FrozenTags.java @@ -0,0 +1,274 @@ +package datadog.trace.bootstrap.instrumentation.api; + +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 + * 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 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. + */ +public final class FrozenTags { + /** No tags. */ + 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) { + 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(); + } + + /** 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 + * SpanPrototype#seedPlan()}: the plan is immutable, so a reader sees null or a complete plan. + */ + private Object applyPlan; + + private FrozenTags(final TagMap tags, final Write[] writes, final boolean replayOnly) { + this.tags = tags; + this.writes = writes; + this.replayOnly = replayOnly; + } + + /** The net result of the recorded writes, frozen. */ + public TagMap tags() { + return tags; + } + + /** Whether nothing was written. */ + public boolean 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. */ + @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(); + private final List writes = new ArrayList<>(); + private boolean replayOnly; + + FrozenTags 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) { + 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; + } + + @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/api/TagStoresTest.java b/internal-api/src/test/java/datadog/trace/api/TagStoresTest.java index 038d414b5e2..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,6 +248,113 @@ void theHashStoreNeverSeedsInBulk() { HashIdStore target = new HashIdStore(null); assertFalse(target.seedAbsent(denseSource(KnownTags.COMPONENT_ID))); assertFalse(target.seedAbsent(new HashIdStore(null))); + assertFalse( + target.seedPartial(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.seedPartial( + 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.seedPartial( + 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.seedPartial( + 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.seedPartial(child, target, TagStoresTest::put), "read-through source"); + assertEquals(1, target.size(), "nothing copied"); } @Test 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..8318f587adb --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/bootstrap/instrumentation/api/FrozenTagsTest.java @@ -0,0 +1,141 @@ +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.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 org.junit.jupiter.api.Test; +import org.mockito.InOrder; + +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 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)); + + 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 + 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(KnownTags.PEER_IPV4_ID, (CharSequence) "10.0.0.1"); + } +}