From 7e5d03899b7313ceaf29063e2e9bcbf881cd226a Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 8 Sep 2026 18:17:00 -0700 Subject: [PATCH 01/12] Fix `Lettuce5MasterReplicaTest` span wait race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace `writer.waitForTraces(1)` with a predicate-based wait for the Redis `SET` span. This avoids relying on `ListWriter`’s cumulative trace count after `writer.clear()` and prevents the assertion from running before the command span is written. --- .../test/java/Lettuce5MasterReplicaTest.java | 28 ++++++++++++------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5MasterReplicaTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5MasterReplicaTest.java index 40d8564e34c..0b1e0a1c3f3 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5MasterReplicaTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5MasterReplicaTest.java @@ -71,10 +71,25 @@ void staticMasterReplicaCommandSpanHasPeerHostname() throws Exception { String result = connection.sync().set("TESTSETKEY", "TESTSETVAL"); assertEquals("OK", result); - writer.waitForTraces(1); + List setSpans = waitForSetSpans(); + assertEquals(1, setSpans.size(), "expected exactly one SET command span"); + DDSpan span = setSpans.get(0); + assertEquals("SET", String.valueOf(span.getResourceName())); + assertEquals("redis-client", String.valueOf(span.getTag(Tags.COMPONENT))); + assertEquals("redis", span.getTag(Tags.DB_TYPE)); + assertNotNull(span.getTag(Tags.PEER_HOSTNAME), "command span should include peer.hostname"); + assertEquals(host, span.getTag(Tags.PEER_HOSTNAME)); + } + + private List waitForSetSpans() { + blockUntilTracesMatch(traces -> !findSetSpans(traces).isEmpty()); + return findSetSpans(writer); + } + + private static List findSetSpans(Iterable> traces) { List setSpans = new ArrayList<>(); - for (List trace : writer) { + for (List trace : traces) { for (DDSpan span : trace) { if ("SET".contentEquals(span.getResourceName()) && "redis-client".equals(String.valueOf(span.getTag(Tags.COMPONENT)))) { @@ -82,14 +97,7 @@ void staticMasterReplicaCommandSpanHasPeerHostname() throws Exception { } } } - - assertEquals(1, setSpans.size(), "expected exactly one SET command span"); - DDSpan span = setSpans.get(0); - assertEquals("SET", String.valueOf(span.getResourceName())); - assertEquals("redis-client", String.valueOf(span.getTag(Tags.COMPONENT))); - assertEquals("redis", span.getTag(Tags.DB_TYPE)); - assertNotNull(span.getTag(Tags.PEER_HOSTNAME), "command span should include peer.hostname"); - assertEquals(host, span.getTag(Tags.PEER_HOSTNAME)); + return setSpans; } @SuppressWarnings("unchecked") From 178d3d5e3663af52739e4309069c810970c31d7e Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 8 Sep 2026 18:26:49 -0700 Subject: [PATCH 02/12] Reproduce missing peer.hostname with Lettuce running with redis cluster --- .../src/test/java/Lettuce5ClusterTest.java | 115 ++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java new file mode 100644 index 00000000000..4ce255a28a2 --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -0,0 +1,115 @@ +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import com.redis.testcontainers.RedisClusterContainer; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.core.DDSpan; +import datadog.trace.test.util.PollingConditions; +import io.lettuce.core.RedisFuture; +import io.lettuce.core.RedisURI; +import io.lettuce.core.cluster.ClusterClientOptions; +import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.SlotHash; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.cluster.models.partitions.RedisClusterNode; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.utility.DockerImageName; + +class Lettuce5ClusterTest extends AbstractInstrumentationTest { + private static final String TEST_SET_KEY = "TESTSETKEY"; + private static final String TEST_SET_VALUE = "TESTSETVAL"; + + private RedisClusterContainer redisCluster; + private RedisClusterClient redisClient; + private StatefulRedisClusterConnection connection; + + @BeforeEach + void setUpRedis() throws Exception { + redisCluster = + new RedisClusterContainer( + // RedisClusterContainer is built around this preconfigured cluster image. + DockerImageName.parse("grokzen/redis-cluster:6.2.1")); + redisCluster.start(); + + redisClient = RedisClusterClient.create(RedisURI.create(redisCluster.getRedisURI())); + redisClient.setOptions(ClusterClientOptions.builder().build()); + connection = redisClient.connect(); + new PollingConditions(30) + .delay(0.5) + .eventually( + () -> + assertEquals( + "OK", + connection.sync().set("DD_CLUSTER_READY", "1"), + "Redis cluster did not become ready")); + + tracer.flush(); + writer.clear(); + } + + @AfterEach + void cleanUpRedis() { + if (connection != null) { + connection.close(); + } + + if (redisClient != null) { + redisClient.shutdown(5, 10, TimeUnit.SECONDS); + } + + if (redisCluster != null) { + redisCluster.stop(); + } + } + + @Test + void clusterCommandSpanHasPeerHostname() throws Exception { + String result = connection.sync().set(TEST_SET_KEY, TEST_SET_VALUE); + + assertEquals("OK", result); + assertSetSpanHasPeerHostname(); + } + + @Test + void asyncClusterCommandSpanHasPeerHostname() throws Exception { + RedisFuture redisFuture = connection.async().set(TEST_SET_KEY, TEST_SET_VALUE); + String result = redisFuture.get(3, TimeUnit.SECONDS); + + assertEquals("OK", result); + assertSetSpanHasPeerHostname(); + } + + private void assertSetSpanHasPeerHostname() throws Exception { + writer.waitForTraces(1); + + RedisClusterNode expectedNode = + connection.getPartitions().getPartitionBySlot(SlotHash.getSlot(TEST_SET_KEY)); + assertNotNull(expectedNode, "expected a cluster node for the command key slot"); + + List setSpans = new ArrayList<>(); + for (List trace : writer) { + for (DDSpan span : trace) { + if ("SET".contentEquals(span.getResourceName()) + && "redis-client".equals(String.valueOf(span.getTag(Tags.COMPONENT)))) { + setSpans.add(span); + } + } + } + + assertFalse(setSpans.isEmpty(), "expected at least one SET command span"); + for (DDSpan span : setSpans) { + assertEquals("SET", String.valueOf(span.getResourceName())); + assertEquals("redis-client", String.valueOf(span.getTag(Tags.COMPONENT))); + assertEquals("redis", span.getTag(Tags.DB_TYPE)); + assertNotNull(span.getTag(Tags.PEER_HOSTNAME), "command span should include peer.hostname"); + assertEquals(expectedNode.getUri().getHost(), span.getTag(Tags.PEER_HOSTNAME)); + } + } +} From 146a01b9e8c5d0d6961327de0cc83e2b53e1b518 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 8 Sep 2026 20:12:11 -0700 Subject: [PATCH 03/12] Fix missing peer.hostname on Lettuce cluster spans --- ...sterConnectionProviderInstrumentation.java | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java new file mode 100644 index 00000000000..e8b7837db4a --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java @@ -0,0 +1,86 @@ +package datadog.trace.instrumentation.lettuce5; + +import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan; +import static datadog.trace.instrumentation.lettuce5.LettuceClientDecorator.DECORATE; +import static net.bytebuddy.matcher.ElementMatchers.isMethod; +import static net.bytebuddy.matcher.ElementMatchers.isPublic; +import static net.bytebuddy.matcher.ElementMatchers.returns; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import com.google.auto.service.AutoService; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.agent.tooling.InstrumenterModule; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import io.lettuce.core.RedisURI; +import io.lettuce.core.cluster.models.partitions.Partitions; +import io.lettuce.core.cluster.models.partitions.RedisClusterNode; +import net.bytebuddy.asm.Advice; + +/** + * Decorates Redis cluster command spans with the node selected for the command key slot. + * + *

Cluster command spans are started before Lettuce resolves the slot owner. This hooks the + * provider lookup that has both the routed slot and the current cluster partitions, then tags the + * active span with the {@link RedisURI} of the node serving that slot. + * + *

This complements the connection context store, which captures RedisURI per physical connection + * but not this per-command slot decision. + */ +@AutoService(InstrumenterModule.class) +public class PooledClusterConnectionProviderInstrumentation extends InstrumenterModule.Tracing + implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { + + public PooledClusterConnectionProviderInstrumentation() { + super("lettuce", "lettuce-5"); + } + + @Override + public String instrumentedType() { + return "io.lettuce.core.cluster.PooledClusterConnectionProvider"; + } + + @Override + public String[] helperClassNames() { + return new String[] { + packageName + ".LettuceClientDecorator", packageName + ".MasterReplicaConnectionHelper", + }; + } + + @Override + public void methodAdvice(MethodTransformer transformer) { + transformer.applyAdvice( + isMethod() + .and(isPublic()) + // Synchronous getConnection delegates here after resolving the command slot. + .and(named("getConnectionAsync")) + .and(takesArguments(2)) + .and(takesArgument(1, int.class)) + .and(returns(named("java.util.concurrent.CompletableFuture"))), + PooledClusterConnectionProviderInstrumentation.class.getName() + "$ConnectionAdvice"); + } + + public static class ConnectionAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class) + public static void onEnter( + @Advice.FieldValue("partitions") final Partitions partitions, + @Advice.Argument(1) final int slot) { + final AgentSpan span = activeSpan(); + if (!MasterReplicaConnectionHelper.isRedisClientSpan(span) || partitions == null) { + return; + } + + final RedisClusterNode node = partitions.getPartitionBySlot(slot); + if (node == null) { + return; + } + + final RedisURI redisURI = node.getUri(); + if (redisURI != null) { + DECORATE.onConnection(span, redisURI); + } + } + } +} From e3a7c0cc7cfc550b7bd60e783e7e558d4fc28b02 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 09:41:58 -0700 Subject: [PATCH 04/12] Fixed the CI failure in Lettuce5ClusterTest Root cause: the test used `grokzen/redis-cluster:6.2.1`, and Docker now returns `NotFound` for that tag. Docker Hub currently lists newer `6.2.x` tags such as `6.2.14` --- .../lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java index 4ce255a28a2..fa18e6a2b6a 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -35,7 +35,7 @@ void setUpRedis() throws Exception { redisCluster = new RedisClusterContainer( // RedisClusterContainer is built around this preconfigured cluster image. - DockerImageName.parse("grokzen/redis-cluster:6.2.1")); + DockerImageName.parse("grokzen/redis-cluster:6.2.14")); redisCluster.start(); redisClient = RedisClusterClient.create(RedisURI.create(redisCluster.getRedisURI())); From 13f4e923646a46020e76a4491558a5f55676d298 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 10:48:05 -0700 Subject: [PATCH 05/12] Fixed Lettuce5 muzzle test --- .../PooledClusterConnectionProviderInstrumentation.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java index e8b7837db4a..267bb27d145 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java @@ -44,7 +44,9 @@ public String instrumentedType() { @Override public String[] helperClassNames() { return new String[] { - packageName + ".LettuceClientDecorator", packageName + ".MasterReplicaConnectionHelper", + packageName + ".LettuceClientDecorator", + packageName + ".MasterReplicaConnectionHelper", + packageName + ".LettuceInstrumentationUtil" }; } From 58176d0cce7d0f9ab51e201a3ea28919f81a5f70 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 12:27:15 -0700 Subject: [PATCH 06/12] Use the official Redis image to bootstrap a single-node cluster in Lettuce5ClusterTest available in the CI registry mirror. --- .../src/test/java/Lettuce5ClusterTest.java | 47 +++++++++++++++---- 1 file changed, 38 insertions(+), 9 deletions(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java index fa18e6a2b6a..c504b305075 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -1,8 +1,9 @@ +import static datadog.trace.agent.test.utils.PortUtils.randomOpenPort; +import static java.util.Collections.singletonList; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; -import com.redis.testcontainers.RedisClusterContainer; import datadog.trace.agent.test.AbstractInstrumentationTest; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.core.DDSpan; @@ -20,25 +21,53 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.testcontainers.utility.DockerImageName; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; class Lettuce5ClusterTest extends AbstractInstrumentationTest { private static final String TEST_SET_KEY = "TESTSETKEY"; private static final String TEST_SET_VALUE = "TESTSETVAL"; + private static final int REDIS_CLUSTER_CONTAINER_PORT = 7000; - private RedisClusterContainer redisCluster; + private GenericContainer redisCluster; private RedisClusterClient redisClient; private StatefulRedisClusterConnection connection; @BeforeEach void setUpRedis() throws Exception { - redisCluster = - new RedisClusterContainer( - // RedisClusterContainer is built around this preconfigured cluster image. - DockerImageName.parse("grokzen/redis-cluster:6.2.14")); + int redisClusterHostPort = randomOpenPort(); + // Redis cluster discovery returns the announced node port, so the host-side port must be + // stable. + redisCluster = new GenericContainer<>("redis:6.2.6"); + redisCluster.setPortBindings( + singletonList(redisClusterHostPort + ":" + REDIS_CLUSTER_CONTAINER_PORT)); + redisCluster + .withExposedPorts(REDIS_CLUSTER_CONTAINER_PORT) + .withCommand( + "sh", + "-c", + "redis-server --port " + + REDIS_CLUSTER_CONTAINER_PORT + + " --cluster-enabled yes" + + " --cluster-node-timeout 5000 --appendonly no --protected-mode no" + + " --cluster-announce-ip 127.0.0.1 --cluster-announce-port " + + redisClusterHostPort + + " & pid=$!; " + + "until redis-cli -p " + + REDIS_CLUSTER_CONTAINER_PORT + + " ping; do sleep 0.1; done; " + + "redis-cli -p " + + REDIS_CLUSTER_CONTAINER_PORT + + " cluster addslots $(seq 0 16383) && echo CLUSTER_READY; " + + "wait $pid") + .waitingFor(Wait.forLogMessage(".*CLUSTER_READY.*\\n", 1)); redisCluster.start(); - redisClient = RedisClusterClient.create(RedisURI.create(redisCluster.getRedisURI())); + RedisURI redisURI = + RedisURI.Builder.redis( + redisCluster.getHost(), redisCluster.getMappedPort(REDIS_CLUSTER_CONTAINER_PORT)) + .build(); + redisClient = RedisClusterClient.create(redisURI); redisClient.setOptions(ClusterClientOptions.builder().build()); connection = redisClient.connect(); new PollingConditions(30) @@ -65,7 +94,7 @@ void cleanUpRedis() { } if (redisCluster != null) { - redisCluster.stop(); + redisCluster.close(); } } From e0cb61427663db265714879fd5c338339db22a26 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 14:21:12 -0700 Subject: [PATCH 07/12] Fix Lettuce cluster peer tags for replica reads Decorate Lettuce cluster command spans from the physical connection selected by the cluster provider instead of deriving peer tags from the slot-owning master. When ReadFrom routes a cluster read to a replica, getConnectionAsync(READ, slot) can return a replica connection while the slot partition still points to the master. Track RedisURI on the selected write/read connection futures and use that connection context when decorating the active command span. Expand Lettuce5ClusterTest to run a single-shard master/replica Redis Cluster, verify replica-routed reads use the replica peer tags, and keep random exposed ports with room for Redis cluster bus ports. --- .../lettuce5/ConnectionContextFunction.java | 26 +++ ...sterConnectionProviderInstrumentation.java | 88 ++++++-- .../src/test/java/Lettuce5ClusterTest.java | 203 +++++++++++++++--- 3 files changed, 278 insertions(+), 39 deletions(-) create mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java new file mode 100644 index 00000000000..17b4ea93d86 --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java @@ -0,0 +1,26 @@ +package datadog.trace.instrumentation.lettuce5; + +import datadog.trace.bootstrap.ContextStore; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import java.util.function.Function; + +public class ConnectionContextFunction implements Function { + + private final RedisURI redisURI; + private final ContextStore contextStore; + + public ConnectionContextFunction( + RedisURI redisURI, ContextStore contextStore) { + this.redisURI = redisURI; + this.contextStore = contextStore; + } + + @Override + public StatefulConnection apply(StatefulConnection statefulConnection) { + if (statefulConnection != null) { + contextStore.put(statefulConnection, redisURI); + } + return statefulConnection; + } +} diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java index 267bb27d145..4a65518990b 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java @@ -2,7 +2,6 @@ import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan; -import static datadog.trace.instrumentation.lettuce5.LettuceClientDecorator.DECORATE; import static net.bytebuddy.matcher.ElementMatchers.isMethod; import static net.bytebuddy.matcher.ElementMatchers.isPublic; import static net.bytebuddy.matcher.ElementMatchers.returns; @@ -12,21 +11,26 @@ import com.google.auto.service.AutoService; import datadog.trace.agent.tooling.Instrumenter; import datadog.trace.agent.tooling.InstrumenterModule; +import datadog.trace.bootstrap.ContextStore; +import datadog.trace.bootstrap.InstrumentationContext; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; import io.lettuce.core.cluster.models.partitions.Partitions; import io.lettuce.core.cluster.models.partitions.RedisClusterNode; +import io.lettuce.core.models.role.RedisNodeDescription; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; import net.bytebuddy.asm.Advice; /** - * Decorates Redis cluster command spans with the node selected for the command key slot. + * Decorates Redis cluster command spans with the physical node selected for the command key slot. * - *

Cluster command spans are started before Lettuce resolves the slot owner. This hooks the - * provider lookup that has both the routed slot and the current cluster partitions, then tags the - * active span with the {@link RedisURI} of the node serving that slot. - * - *

This complements the connection context store, which captures RedisURI per physical connection - * but not this per-command slot decision. + *

Cluster command spans are started before Lettuce resolves the slot owner and applies {@code + * ReadFrom}. This tracks the {@link RedisURI} of physical cluster node connections, then tags the + * active command span when Lettuce returns the selected connection. */ @AutoService(InstrumenterModule.class) public class PooledClusterConnectionProviderInstrumentation extends InstrumenterModule.Tracing @@ -41,11 +45,18 @@ public String instrumentedType() { return "io.lettuce.core.cluster.PooledClusterConnectionProvider"; } + @Override + public Map contextStore() { + return Collections.singletonMap( + "io.lettuce.core.api.StatefulConnection", "io.lettuce.core.RedisURI"); + } + @Override public String[] helperClassNames() { return new String[] { packageName + ".LettuceClientDecorator", packageName + ".MasterReplicaConnectionHelper", + packageName + ".ConnectionContextFunction", packageName + ".LettuceInstrumentationUtil" }; } @@ -61,16 +72,39 @@ public void methodAdvice(MethodTransformer transformer) { .and(takesArgument(1, int.class)) .and(returns(named("java.util.concurrent.CompletableFuture"))), PooledClusterConnectionProviderInstrumentation.class.getName() + "$ConnectionAdvice"); + transformer.applyAdvice( + isMethod().and(named("getWriteConnection")).and(takesArguments(1)), + PooledClusterConnectionProviderInstrumentation.class.getName() + "$WriteConnectionAdvice"); + transformer.applyAdvice( + isMethod().and(named("getReadFromConnections")).and(takesArguments(1)), + PooledClusterConnectionProviderInstrumentation.class.getName() + "$ReadConnectionAdvice"); } public static class ConnectionAdvice { - @Advice.OnMethodEnter(suppress = Throwable.class) - public static void onEnter( - @Advice.FieldValue("partitions") final Partitions partitions, - @Advice.Argument(1) final int slot) { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.Return final CompletableFuture connectionFuture) { final AgentSpan span = activeSpan(); - if (!MasterReplicaConnectionHelper.isRedisClientSpan(span) || partitions == null) { + if (!MasterReplicaConnectionHelper.isRedisClientSpan(span) || connectionFuture == null) { + return; + } + + connectionFuture.whenComplete( + MasterReplicaConnectionHelper.onConnectionComplete( + span, InstrumentationContext.get(StatefulConnection.class, RedisURI.class))); + } + } + + public static class WriteConnectionAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.FieldValue("partitions") final Partitions partitions, + @Advice.Argument(0) final int slot, + @Advice.Return(readOnly = false) + CompletableFuture connectionFuture) { + if (partitions == null || connectionFuture == null) { return; } @@ -81,7 +115,33 @@ public static void onEnter( final RedisURI redisURI = node.getUri(); if (redisURI != null) { - DECORATE.onConnection(span, redisURI); + connectionFuture = + connectionFuture.thenApply( + new ConnectionContextFunction( + redisURI, + InstrumentationContext.get(StatefulConnection.class, RedisURI.class))); + } + } + } + + public static class ReadConnectionAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.Argument(0) final List selection, + @Advice.Return final CompletableFuture[] connectionFutures) { + if (selection == null || connectionFutures == null) { + return; + } + + final ContextStore contextStore = + InstrumentationContext.get(StatefulConnection.class, RedisURI.class); + for (int i = 0; i < selection.size() && i < connectionFutures.length; i++) { + final RedisURI redisURI = selection.get(i).getUri(); + if (redisURI != null && connectionFutures[i] != null) { + connectionFutures[i] = + connectionFutures[i].thenApply(new ConnectionContextFunction(redisURI, contextStore)); + } } } } diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java index c504b305075..2ef2e961f9c 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -1,13 +1,15 @@ import static datadog.trace.agent.test.utils.PortUtils.randomOpenPort; -import static java.util.Collections.singletonList; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.fail; import datadog.trace.agent.test.AbstractInstrumentationTest; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.core.DDSpan; import datadog.trace.test.util.PollingConditions; +import io.lettuce.core.ReadFrom; import io.lettuce.core.RedisFuture; import io.lettuce.core.RedisURI; import io.lettuce.core.cluster.ClusterClientOptions; @@ -16,6 +18,7 @@ import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.cluster.models.partitions.RedisClusterNode; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.AfterEach; @@ -27,46 +30,43 @@ class Lettuce5ClusterTest extends AbstractInstrumentationTest { private static final String TEST_SET_KEY = "TESTSETKEY"; private static final String TEST_SET_VALUE = "TESTSETVAL"; - private static final int REDIS_CLUSTER_CONTAINER_PORT = 7000; + private static final int MAX_TCP_PORT = 65535; + private static final int CLUSTER_BUS_PORT_OFFSET = 10000; + private static final int MAX_CLUSTER_DATA_PORT = MAX_TCP_PORT - CLUSTER_BUS_PORT_OFFSET; private GenericContainer redisCluster; private RedisClusterClient redisClient; private StatefulRedisClusterConnection connection; + private int redisClusterMasterPort; + private int redisClusterReplicaPort; @BeforeEach void setUpRedis() throws Exception { - int redisClusterHostPort = randomOpenPort(); + redisClusterMasterPort = randomClusterPort(MAX_CLUSTER_DATA_PORT); + redisClusterReplicaPort = randomClusterPort(MAX_CLUSTER_DATA_PORT); + while (redisClusterMasterPort == redisClusterReplicaPort + || redisClusterMasterPort + CLUSTER_BUS_PORT_OFFSET == redisClusterReplicaPort + || redisClusterReplicaPort + CLUSTER_BUS_PORT_OFFSET == redisClusterMasterPort) { + redisClusterReplicaPort = randomClusterPort(MAX_CLUSTER_DATA_PORT); + } + // Redis cluster discovery returns the announced node port, so the host-side port must be - // stable. + // stable. Use the same random ports inside the container so cluster nodes can also reach each + // other at their announced addresses. redisCluster = new GenericContainer<>("redis:6.2.6"); redisCluster.setPortBindings( - singletonList(redisClusterHostPort + ":" + REDIS_CLUSTER_CONTAINER_PORT)); + Arrays.asList( + redisClusterMasterPort + ":" + redisClusterMasterPort, + redisClusterReplicaPort + ":" + redisClusterReplicaPort)); redisCluster - .withExposedPorts(REDIS_CLUSTER_CONTAINER_PORT) + .withExposedPorts(redisClusterMasterPort, redisClusterReplicaPort) .withCommand( - "sh", - "-c", - "redis-server --port " - + REDIS_CLUSTER_CONTAINER_PORT - + " --cluster-enabled yes" - + " --cluster-node-timeout 5000 --appendonly no --protected-mode no" - + " --cluster-announce-ip 127.0.0.1 --cluster-announce-port " - + redisClusterHostPort - + " & pid=$!; " - + "until redis-cli -p " - + REDIS_CLUSTER_CONTAINER_PORT - + " ping; do sleep 0.1; done; " - + "redis-cli -p " - + REDIS_CLUSTER_CONTAINER_PORT - + " cluster addslots $(seq 0 16383) && echo CLUSTER_READY; " - + "wait $pid") + "sh", "-c", redisClusterCommand(redisClusterMasterPort, redisClusterReplicaPort)) .waitingFor(Wait.forLogMessage(".*CLUSTER_READY.*\\n", 1)); redisCluster.start(); RedisURI redisURI = - RedisURI.Builder.redis( - redisCluster.getHost(), redisCluster.getMappedPort(REDIS_CLUSTER_CONTAINER_PORT)) - .build(); + RedisURI.Builder.redis(redisCluster.getHost(), redisClusterMasterPort).build(); redisClient = RedisClusterClient.create(redisURI); redisClient.setOptions(ClusterClientOptions.builder().build()); connection = redisClient.connect(); @@ -115,6 +115,24 @@ void asyncClusterCommandSpanHasPeerHostname() throws Exception { assertSetSpanHasPeerHostname(); } + @Test + void clusterReadCommandSpanUsesReplicaPeerWithReadFromReplica() throws Exception { + assertEquals("OK", connection.sync().set(TEST_SET_KEY, TEST_SET_VALUE)); + connection.setReadFrom(ReadFrom.SLAVE); + new PollingConditions(30) + .delay(0.5) + .eventually(() -> assertEquals(TEST_SET_VALUE, connection.sync().get(TEST_SET_KEY))); + + blockUntilTracesMatch(traces -> !findCommandSpans(traces, "GET").isEmpty()); + tracer.flush(); + writer.clear(); + + String result = connection.sync().get(TEST_SET_KEY); + + assertEquals(TEST_SET_VALUE, result); + assertGetSpanHasReplicaPeer(); + } + private void assertSetSpanHasPeerHostname() throws Exception { writer.waitForTraces(1); @@ -141,4 +159,139 @@ private void assertSetSpanHasPeerHostname() throws Exception { assertEquals(expectedNode.getUri().getHost(), span.getTag(Tags.PEER_HOSTNAME)); } } + + private void assertGetSpanHasReplicaPeer() { + blockUntilTracesMatch(traces -> !findCommandSpans(traces, "GET").isEmpty()); + + RedisClusterNode master = + connection.getPartitions().getPartitionBySlot(SlotHash.getSlot(TEST_SET_KEY)); + assertNotNull(master, "expected a cluster master node for the command key slot"); + + RedisClusterNode replica = findReplicaOf(master); + assertNotNull(replica, "expected a replica for the command key slot"); + assertNotEquals( + master.getUri().getPort(), + replica.getUri().getPort(), + "test must use different master and replica endpoints"); + + List getSpans = findCommandSpans(writer, "GET"); + assertFalse(getSpans.isEmpty(), "expected at least one GET command span"); + for (DDSpan span : getSpans) { + assertEquals("GET", String.valueOf(span.getResourceName())); + assertEquals("redis-client", String.valueOf(span.getTag(Tags.COMPONENT))); + assertEquals("redis", span.getTag(Tags.DB_TYPE)); + assertEquals(replica.getUri().getPort(), span.getTag(Tags.PEER_PORT)); + assertEquals(replica.getUri().getHost(), span.getTag(Tags.PEER_HOSTNAME)); + } + } + + private RedisClusterNode findReplicaOf(RedisClusterNode master) { + for (RedisClusterNode node : connection.getPartitions()) { + if (master.getNodeId().equals(node.getSlaveOf())) { + return node; + } + } + fail("No replica found for master " + master); + return null; + } + + private static List findCommandSpans(Iterable> traces, String command) { + List commandSpans = new ArrayList<>(); + for (List trace : traces) { + for (DDSpan span : trace) { + if (command.contentEquals(span.getResourceName()) + && "redis-client".equals(String.valueOf(span.getTag(Tags.COMPONENT)))) { + commandSpans.add(span); + } + } + } + return commandSpans; + } + + private static int randomClusterPort(int maxPort) { + int port = randomOpenPort(); + while (port > maxPort) { + port = randomOpenPort(); + } + return port; + } + + private static String redisClusterCommand(int masterPort, int replicaPort) { + return "set -e; " + + "mkdir -p /tmp/redis-cluster; " + // Start the slot-owning master on its announced data and cluster bus ports. + + "redis-server --port " + + masterPort + + " --dir /tmp/redis-cluster --cluster-enabled yes --cluster-config-file nodes-" + + masterPort + + ".conf --cluster-node-timeout 5000 --appendonly no --protected-mode no" + + " --cluster-announce-ip 127.0.0.1 --cluster-announce-port " + + masterPort + + " --cluster-announce-bus-port " + + (masterPort + CLUSTER_BUS_PORT_OFFSET) + + " --daemonize yes; " + // Start the replica on its own announced data and cluster bus ports. + + "redis-server --port " + + replicaPort + + " --dir /tmp/redis-cluster --cluster-enabled yes --cluster-config-file nodes-" + + replicaPort + + ".conf --cluster-node-timeout 5000 --appendonly no --protected-mode no" + + " --cluster-announce-ip 127.0.0.1 --cluster-announce-port " + + replicaPort + + " --cluster-announce-bus-port " + + (replicaPort + CLUSTER_BUS_PORT_OFFSET) + + " --daemonize yes; " + // Wait until both Redis server processes accept commands. + + "until redis-cli -p " + + masterPort + + " ping; do sleep 0.1; done; " + + "until redis-cli -p " + + replicaPort + + " ping; do sleep 0.1; done; " + // Assign every slot to one master so the test cluster is valid with a single shard. + + "redis-cli -p " + + masterPort + + " cluster addslots $(seq 0 16383); " + // Introduce the replica node to the master's cluster view. + + "redis-cli -p " + + replicaPort + + " cluster meet 127.0.0.1 " + + masterPort + + "; " + // Capture stable node IDs needed for replication checks. + + "master_id=$(redis-cli -p " + + masterPort + + " cluster myid); " + + "replica_id=$(redis-cli -p " + + replicaPort + + " cluster myid); " + // Wait until the replica sees the master before requesting replication. + + "until redis-cli -p " + + replicaPort + + " cluster nodes | grep \"$master_id\"; do sleep 0.1; done; " + // Convert the second node into a replica of the slot-owning master. + + "redis-cli -p " + + replicaPort + + " cluster replicate \"$master_id\"; " + // Wait until the master's cluster view records the replica relationship. + + "until redis-cli -p " + + masterPort + + " cluster nodes | grep \"$replica_id\" | grep \"$master_id\" | grep -q slave; do sleep 0.1; done; " + // Wait until the replica's local cluster view records its replica role. + + "until redis-cli -p " + + replicaPort + + " cluster nodes | grep \"$replica_id\" | grep \"$master_id\" | grep -q myself,slave; do sleep 0.1; done; " + // Wait until Redis reports the process role as replica. + + "until redis-cli -p " + + replicaPort + + " role | grep -q slave; do sleep 0.1; done; " + // Wait until the cluster is usable before releasing the Testcontainers wait strategy. + + "until redis-cli -p " + + masterPort + + " cluster info | grep -q cluster_state:ok; do sleep 0.1; done; " + // Signal readiness to the Java test. + + "echo CLUSTER_READY; " + // Keep the container alive for the duration of the test. + + "tail -f /dev/null"; + } } From 726c6180ef1e832b82045f0bc9e38411b0b1fdfe Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 15:17:45 -0700 Subject: [PATCH 08/12] Fixed the Lettuce 6.2 failure. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cause: Lettuce 6.2’s replica read selection requires the replica partition metadata to have a non-zero replication offset. The test created the replica correctly, but the client could still have stale topology metadata when `ReadFrom.SLAVE` was enabled, causing `PartitionSelectorException`. --- .../lettuce-5.0/src/test/java/Lettuce5ClusterTest.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java index 2ef2e961f9c..e4015175229 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -121,7 +121,11 @@ void clusterReadCommandSpanUsesReplicaPeerWithReadFromReplica() throws Exception connection.setReadFrom(ReadFrom.SLAVE); new PollingConditions(30) .delay(0.5) - .eventually(() -> assertEquals(TEST_SET_VALUE, connection.sync().get(TEST_SET_KEY))); + .eventually( + () -> { + redisClient.reloadPartitions(); + assertEquals(TEST_SET_VALUE, connection.sync().get(TEST_SET_KEY)); + }); blockUntilTracesMatch(traces -> !findCommandSpans(traces, "GET").isEmpty()); tracer.flush(); From aa07467272327721de31905667126ae496d66302 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 18:27:05 -0700 Subject: [PATCH 09/12] Populate Lettuce cluster context at node connection creation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move `StatefulConnection -> RedisURI` context population from per-command pooled cluster connection lookups to `RedisClusterClient.connectToNodeAsync`, deriving the URI from Lettuce’s resolved `ConnectionFuture` remote address. This removes the hot-path `thenApply` wrappers from cluster read/write selection while preserving command span decoration for the selected physical node. Also add a completed-future fast path for cluster and master/replica span decoration callbacks. --- .../ClusterConnectionContextFunction.java | 53 ++++++++++++++ .../lettuce5/ConnectionContextFunction.java | 26 ------- .../MasterReplicaConnectionHelper.java | 15 ++++ ...licaConnectionProviderInstrumentation.java | 7 +- ...sterConnectionProviderInstrumentation.java | 69 ++---------------- .../RedisClusterClientInstrumentation.java | 71 +++++++++++++++++++ 6 files changed, 147 insertions(+), 94 deletions(-) create mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ClusterConnectionContextFunction.java delete mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java create mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/RedisClusterClientInstrumentation.java diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ClusterConnectionContextFunction.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ClusterConnectionContextFunction.java new file mode 100644 index 00000000000..cc6b1fb25e5 --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ClusterConnectionContextFunction.java @@ -0,0 +1,53 @@ +package datadog.trace.instrumentation.lettuce5; + +import datadog.trace.bootstrap.ContextStore; +import io.lettuce.core.ConnectionFuture; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.function.Function; + +public class ClusterConnectionContextFunction + implements Function { + + private final ConnectionFuture connectionFuture; + private final ContextStore contextStore; + + public ClusterConnectionContextFunction( + final ConnectionFuture connectionFuture, + final ContextStore contextStore) { + this.connectionFuture = connectionFuture; + this.contextStore = contextStore; + } + + @Override + public T apply(final T connection) { + if (connection == null) { + return null; + } + + try { + final RedisURI connectionURI = redisUriFromConnectionFuture(); + if (connectionURI != null) { + contextStore.put(connection, connectionURI); + } + } catch (final Throwable ignored) { + } + return connection; + } + + private RedisURI redisUriFromConnectionFuture() { + if (connectionFuture == null) { + return null; + } + + final SocketAddress socketAddress = connectionFuture.getRemoteAddress(); + if (socketAddress instanceof InetSocketAddress) { + final InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress; + return RedisURI.create(inetSocketAddress.getHostString(), inetSocketAddress.getPort()); + } + + return null; + } +} diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java deleted file mode 100644 index 17b4ea93d86..00000000000 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/ConnectionContextFunction.java +++ /dev/null @@ -1,26 +0,0 @@ -package datadog.trace.instrumentation.lettuce5; - -import datadog.trace.bootstrap.ContextStore; -import io.lettuce.core.RedisURI; -import io.lettuce.core.api.StatefulConnection; -import java.util.function.Function; - -public class ConnectionContextFunction implements Function { - - private final RedisURI redisURI; - private final ContextStore contextStore; - - public ConnectionContextFunction( - RedisURI redisURI, ContextStore contextStore) { - this.redisURI = redisURI; - this.contextStore = contextStore; - } - - @Override - public StatefulConnection apply(StatefulConnection statefulConnection) { - if (statefulConnection != null) { - contextStore.put(statefulConnection, redisURI); - } - return statefulConnection; - } -} diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java index 39261f49eb1..5aa643a4ca6 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java @@ -7,6 +7,7 @@ import datadog.trace.bootstrap.instrumentation.api.Tags; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulConnection; +import java.util.concurrent.CompletableFuture; import java.util.function.BiConsumer; public final class MasterReplicaConnectionHelper { @@ -35,4 +36,18 @@ public static BiConsumer onConnectionComplete( final AgentSpan span, final ContextStore contextStore) { return (connection, _throwable) -> onConnection(span, connection, contextStore); } + + public static void onConnectionFuture( + final AgentSpan span, + final CompletableFuture connectionFuture, + final ContextStore contextStore) { + if (connectionFuture.isDone() + && !connectionFuture.isCompletedExceptionally() + && !connectionFuture.isCancelled()) { + onConnection(span, connectionFuture.getNow(null), contextStore); + return; + } + + connectionFuture.whenComplete(onConnectionComplete(span, contextStore)); + } } diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java index e0f93f7bc20..f2e4a047c73 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java @@ -105,9 +105,10 @@ public static void onExit( return; } - connectionFuture.whenComplete( - MasterReplicaConnectionHelper.onConnectionComplete( - span, InstrumentationContext.get(StatefulConnection.class, RedisURI.class))); + MasterReplicaConnectionHelper.onConnectionFuture( + span, + connectionFuture, + InstrumentationContext.get(StatefulConnection.class, RedisURI.class)); } } } diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java index 4a65518990b..ec4414c8ee5 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java @@ -11,16 +11,11 @@ import com.google.auto.service.AutoService; import datadog.trace.agent.tooling.Instrumenter; import datadog.trace.agent.tooling.InstrumenterModule; -import datadog.trace.bootstrap.ContextStore; import datadog.trace.bootstrap.InstrumentationContext; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulConnection; -import io.lettuce.core.cluster.models.partitions.Partitions; -import io.lettuce.core.cluster.models.partitions.RedisClusterNode; -import io.lettuce.core.models.role.RedisNodeDescription; import java.util.Collections; -import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import net.bytebuddy.asm.Advice; @@ -56,7 +51,6 @@ public String[] helperClassNames() { return new String[] { packageName + ".LettuceClientDecorator", packageName + ".MasterReplicaConnectionHelper", - packageName + ".ConnectionContextFunction", packageName + ".LettuceInstrumentationUtil" }; } @@ -72,12 +66,6 @@ public void methodAdvice(MethodTransformer transformer) { .and(takesArgument(1, int.class)) .and(returns(named("java.util.concurrent.CompletableFuture"))), PooledClusterConnectionProviderInstrumentation.class.getName() + "$ConnectionAdvice"); - transformer.applyAdvice( - isMethod().and(named("getWriteConnection")).and(takesArguments(1)), - PooledClusterConnectionProviderInstrumentation.class.getName() + "$WriteConnectionAdvice"); - transformer.applyAdvice( - isMethod().and(named("getReadFromConnections")).and(takesArguments(1)), - PooledClusterConnectionProviderInstrumentation.class.getName() + "$ReadConnectionAdvice"); } public static class ConnectionAdvice { @@ -90,59 +78,10 @@ public static void onExit( return; } - connectionFuture.whenComplete( - MasterReplicaConnectionHelper.onConnectionComplete( - span, InstrumentationContext.get(StatefulConnection.class, RedisURI.class))); - } - } - - public static class WriteConnectionAdvice { - - @Advice.OnMethodExit(suppress = Throwable.class) - public static void onExit( - @Advice.FieldValue("partitions") final Partitions partitions, - @Advice.Argument(0) final int slot, - @Advice.Return(readOnly = false) - CompletableFuture connectionFuture) { - if (partitions == null || connectionFuture == null) { - return; - } - - final RedisClusterNode node = partitions.getPartitionBySlot(slot); - if (node == null) { - return; - } - - final RedisURI redisURI = node.getUri(); - if (redisURI != null) { - connectionFuture = - connectionFuture.thenApply( - new ConnectionContextFunction( - redisURI, - InstrumentationContext.get(StatefulConnection.class, RedisURI.class))); - } - } - } - - public static class ReadConnectionAdvice { - - @Advice.OnMethodExit(suppress = Throwable.class) - public static void onExit( - @Advice.Argument(0) final List selection, - @Advice.Return final CompletableFuture[] connectionFutures) { - if (selection == null || connectionFutures == null) { - return; - } - - final ContextStore contextStore = - InstrumentationContext.get(StatefulConnection.class, RedisURI.class); - for (int i = 0; i < selection.size() && i < connectionFutures.length; i++) { - final RedisURI redisURI = selection.get(i).getUri(); - if (redisURI != null && connectionFutures[i] != null) { - connectionFutures[i] = - connectionFutures[i].thenApply(new ConnectionContextFunction(redisURI, contextStore)); - } - } + MasterReplicaConnectionHelper.onConnectionFuture( + span, + connectionFuture, + InstrumentationContext.get(StatefulConnection.class, RedisURI.class)); } } } diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/RedisClusterClientInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/RedisClusterClientInstrumentation.java new file mode 100644 index 00000000000..b4d72219af5 --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/RedisClusterClientInstrumentation.java @@ -0,0 +1,71 @@ +package datadog.trace.instrumentation.lettuce5; + +import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.isMethod; +import static net.bytebuddy.matcher.ElementMatchers.returns; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import com.google.auto.service.AutoService; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.agent.tooling.InstrumenterModule; +import datadog.trace.bootstrap.InstrumentationContext; +import io.lettuce.core.ConnectionFuture; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import java.util.Collections; +import java.util.Map; +import net.bytebuddy.asm.Advice; + +@AutoService(InstrumenterModule.class) +public class RedisClusterClientInstrumentation extends InstrumenterModule.Tracing + implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { + + public RedisClusterClientInstrumentation() { + super("lettuce", "lettuce-5"); + } + + @Override + public String instrumentedType() { + return "io.lettuce.core.cluster.RedisClusterClient"; + } + + @Override + public Map contextStore() { + return Collections.singletonMap( + "io.lettuce.core.api.StatefulConnection", "io.lettuce.core.RedisURI"); + } + + @Override + public String[] helperClassNames() { + return new String[] {packageName + ".ClusterConnectionContextFunction"}; + } + + @Override + public void methodAdvice(MethodTransformer transformer) { + transformer.applyAdvice( + isMethod() + .and(named("connectToNodeAsync")) + .and(takesArguments(4)) + .and(takesArgument(1, String.class)) + .and(returns(named("io.lettuce.core.ConnectionFuture"))), + RedisClusterClientInstrumentation.class.getName() + "$ConnectToNodeAdvice"); + } + + public static class ConnectToNodeAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.Return(readOnly = false) ConnectionFuture connectionFuture) { + if (connectionFuture == null) { + return; + } + + connectionFuture = + connectionFuture.thenApply( + new ClusterConnectionContextFunction( + connectionFuture, + InstrumentationContext.get(StatefulConnection.class, RedisURI.class))); + } + } +} From da5c290904c767a70b761762b040066ec99b404c Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 18:41:34 -0700 Subject: [PATCH 10/12] Fix Lettuce cluster SET span wait in tests --- .../src/test/java/Lettuce5ClusterTest.java | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java index e4015175229..e60d6df75b0 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -137,23 +137,14 @@ void clusterReadCommandSpanUsesReplicaPeerWithReadFromReplica() throws Exception assertGetSpanHasReplicaPeer(); } - private void assertSetSpanHasPeerHostname() throws Exception { - writer.waitForTraces(1); + private void assertSetSpanHasPeerHostname() { + blockUntilTracesMatch(traces -> !findCommandSpans(traces, "SET").isEmpty()); RedisClusterNode expectedNode = connection.getPartitions().getPartitionBySlot(SlotHash.getSlot(TEST_SET_KEY)); assertNotNull(expectedNode, "expected a cluster node for the command key slot"); - List setSpans = new ArrayList<>(); - for (List trace : writer) { - for (DDSpan span : trace) { - if ("SET".contentEquals(span.getResourceName()) - && "redis-client".equals(String.valueOf(span.getTag(Tags.COMPONENT)))) { - setSpans.add(span); - } - } - } - + List setSpans = findCommandSpans(writer, "SET"); assertFalse(setSpans.isEmpty(), "expected at least one SET command span"); for (DDSpan span : setSpans) { assertEquals("SET", String.valueOf(span.getResourceName())); From ec2d2983bf10d6041587328e6320bcd22464a4eb Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Wed, 9 Sep 2026 20:23:21 -0700 Subject: [PATCH 11/12] Fix reactive Lettuce cluster peer tagging Activate the stored reactive Redis command span while `RedisSubscription.dispatchCommand()` selects the physical cluster connection, allowing existing connection decoration to add peer tags. Add a reactive cluster regression test covering `peer.hostname` on command spans. --- .../LettuceReactiveClientInstrumentation.java | 6 +++++ .../rx/RedisSubscriptionDispatchAdvice.java | 27 +++++++++++++++++++ .../src/test/java/Lettuce5ClusterTest.java | 8 ++++++ 3 files changed, 41 insertions(+) create mode 100644 dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/rx/RedisSubscriptionDispatchAdvice.java diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/LettuceReactiveClientInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/LettuceReactiveClientInstrumentation.java index 005d3b8bb6f..3240eea74bd 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/LettuceReactiveClientInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/LettuceReactiveClientInstrumentation.java @@ -66,6 +66,7 @@ public String[] helperClassNames() { return new String[] { packageName + ".rx.RedisSubscriptionSubscribeAdvice", packageName + ".rx.RedisSubscriptionSubscribeAdvice$State", + packageName + ".rx.RedisSubscriptionDispatchAdvice", packageName + ".rx.RedisSubscriptionState", packageName + ".LettuceInstrumentationUtil", packageName + ".LettuceClientDecorator", @@ -88,6 +89,11 @@ public Map contextStore() { public void methodAdvice(MethodTransformer transformer) { transformer.applyAdvice( isMethod().and(named("subscribe")), packageName + ".rx.RedisSubscriptionSubscribeAdvice"); + transformer.applyAdvice( + isMethod() + .and(isDeclaredBy(named("io.lettuce.core.RedisPublisher$RedisSubscription"))) + .and(named("dispatchCommand")), + packageName + ".rx.RedisSubscriptionDispatchAdvice"); transformer.applyAdvice( isMethod().and(named("onNext")), packageName + ".rx.RedisSubscriptionAdvanceAdvice"); transformer.applyAdvice( diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/rx/RedisSubscriptionDispatchAdvice.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/rx/RedisSubscriptionDispatchAdvice.java new file mode 100644 index 00000000000..660bf1269ab --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/rx/RedisSubscriptionDispatchAdvice.java @@ -0,0 +1,27 @@ +package datadog.trace.instrumentation.lettuce5.rx; + +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; + +import datadog.trace.bootstrap.InstrumentationContext; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import io.lettuce.core.protocol.RedisCommand; +import net.bytebuddy.asm.Advice; + +public class RedisSubscriptionDispatchAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class) + public static AgentScope beforeDispatch( + @Advice.FieldValue("subscriptionCommand") RedisCommand subscriptionCommand) { + AgentSpan span = + InstrumentationContext.get(RedisCommand.class, AgentSpan.class).get(subscriptionCommand); + return span != null ? activateSpan(span) : null; + } + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) + public static void afterDispatch(@Advice.Enter AgentScope scope) { + if (scope != null) { + scope.close(); + } + } +} diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java index e60d6df75b0..ccefb34e0b7 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -115,6 +115,14 @@ void asyncClusterCommandSpanHasPeerHostname() throws Exception { assertSetSpanHasPeerHostname(); } + @Test + void reactiveClusterCommandSpanHasPeerHostname() { + String result = connection.reactive().set(TEST_SET_KEY, TEST_SET_VALUE).block(); + + assertEquals("OK", result); + assertSetSpanHasPeerHostname(); + } + @Test void clusterReadCommandSpanUsesReplicaPeerWithReadFromReplica() throws Exception { assertEquals("OK", connection.sync().set(TEST_SET_KEY, TEST_SET_VALUE)); From 86193f41d326ba0495396f407b8d9f4626bd8762 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 10 Sep 2026 09:25:38 -0700 Subject: [PATCH 12/12] Sequence Lettuce peer decoration before connection use Return the dependent connection future that performs peer decoration and replace the async advice return values so Lettuce dispatch waits until the selected connection has tagged the active Redis span. Keeps already-completed futures on the synchronous path. --- .../MasterReplicaConnectionHelper.java | 27 +++++++++---------- ...licaConnectionProviderInstrumentation.java | 13 ++++----- ...sterConnectionProviderInstrumentation.java | 13 ++++----- 3 files changed, 27 insertions(+), 26 deletions(-) diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java index 5aa643a4ca6..3a33fedddc7 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java @@ -8,7 +8,6 @@ import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulConnection; import java.util.concurrent.CompletableFuture; -import java.util.function.BiConsumer; public final class MasterReplicaConnectionHelper { @@ -32,22 +31,22 @@ public static void onConnection( } } - public static BiConsumer onConnectionComplete( - final AgentSpan span, final ContextStore contextStore) { - return (connection, _throwable) -> onConnection(span, connection, contextStore); - } - - public static void onConnectionFuture( + public static CompletableFuture onConnectionFuture( final AgentSpan span, - final CompletableFuture connectionFuture, + final CompletableFuture connectionFuture, final ContextStore contextStore) { - if (connectionFuture.isDone() - && !connectionFuture.isCompletedExceptionally() - && !connectionFuture.isCancelled()) { - onConnection(span, connectionFuture.getNow(null), contextStore); - return; + if (connectionFuture.isDone()) { + if (!connectionFuture.isCompletedExceptionally() && !connectionFuture.isCancelled()) { + onConnection(span, connectionFuture.getNow(null), contextStore); + } + return connectionFuture; } - connectionFuture.whenComplete(onConnectionComplete(span, contextStore)); + return connectionFuture.whenComplete( + (connection, throwable) -> { + if (throwable == null) { + onConnection(span, connection, contextStore); + } + }); } } diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java index f2e4a047c73..cfa04bab42b 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionProviderInstrumentation.java @@ -98,17 +98,18 @@ public static void onExit(@Advice.Return final StatefulRedisConnection con public static class AsyncAdvice { @Advice.OnMethodExit(suppress = Throwable.class) - public static void onExit( - @Advice.Return final CompletableFuture connectionFuture) { + public static void onExit( + @Advice.Return(readOnly = false) CompletableFuture connectionFuture) { final AgentSpan span = activeSpan(); if (!MasterReplicaConnectionHelper.isRedisClientSpan(span) || connectionFuture == null) { return; } - MasterReplicaConnectionHelper.onConnectionFuture( - span, - connectionFuture, - InstrumentationContext.get(StatefulConnection.class, RedisURI.class)); + connectionFuture = + MasterReplicaConnectionHelper.onConnectionFuture( + span, + connectionFuture, + InstrumentationContext.get(StatefulConnection.class, RedisURI.class)); } } } diff --git a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java index ec4414c8ee5..d9920b27249 100644 --- a/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java @@ -71,17 +71,18 @@ public void methodAdvice(MethodTransformer transformer) { public static class ConnectionAdvice { @Advice.OnMethodExit(suppress = Throwable.class) - public static void onExit( - @Advice.Return final CompletableFuture connectionFuture) { + public static void onExit( + @Advice.Return(readOnly = false) CompletableFuture connectionFuture) { final AgentSpan span = activeSpan(); if (!MasterReplicaConnectionHelper.isRedisClientSpan(span) || connectionFuture == null) { return; } - MasterReplicaConnectionHelper.onConnectionFuture( - span, - connectionFuture, - InstrumentationContext.get(StatefulConnection.class, RedisURI.class)); + connectionFuture = + MasterReplicaConnectionHelper.onConnectionFuture( + span, + connectionFuture, + InstrumentationContext.get(StatefulConnection.class, RedisURI.class)); } } }