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/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/MasterReplicaConnectionHelper.java b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/MasterReplicaConnectionHelper.java index 39261f49eb1..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 @@ -7,7 +7,7 @@ import datadog.trace.bootstrap.instrumentation.api.Tags; import io.lettuce.core.RedisURI; import io.lettuce.core.api.StatefulConnection; -import java.util.function.BiConsumer; +import java.util.concurrent.CompletableFuture; public final class MasterReplicaConnectionHelper { @@ -31,8 +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 CompletableFuture onConnectionFuture( + final AgentSpan span, + final CompletableFuture connectionFuture, + final ContextStore contextStore) { + if (connectionFuture.isDone()) { + if (!connectionFuture.isCompletedExceptionally() && !connectionFuture.isCancelled()) { + onConnection(span, connectionFuture.getNow(null), contextStore); + } + return connectionFuture; + } + + 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 e0f93f7bc20..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,16 +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; } - connectionFuture.whenComplete( - MasterReplicaConnectionHelper.onConnectionComplete( - span, 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 new file mode 100644 index 00000000000..d9920b27249 --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/main/java/datadog/trace/instrumentation/lettuce5/PooledClusterConnectionProviderInstrumentation.java @@ -0,0 +1,88 @@ +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 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.InstrumentationContext; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import net.bytebuddy.asm.Advice; + +/** + * 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 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 + implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { + + public PooledClusterConnectionProviderInstrumentation() { + super("lettuce", "lettuce-5"); + } + + @Override + 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 + ".LettuceInstrumentationUtil" + }; + } + + @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.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.Return(readOnly = false) CompletableFuture connectionFuture) { + final AgentSpan span = activeSpan(); + if (!MasterReplicaConnectionHelper.isRedisClientSpan(span) || connectionFuture == null) { + return; + } + + 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/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))); + } + } +} 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 new file mode 100644 index 00000000000..ccefb34e0b7 --- /dev/null +++ b/dd-java-agent/instrumentation/lettuce/lettuce-5.0/src/test/java/Lettuce5ClusterTest.java @@ -0,0 +1,300 @@ +import static datadog.trace.agent.test.utils.PortUtils.randomOpenPort; +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; +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.Arrays; +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.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 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 { + 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. 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( + Arrays.asList( + redisClusterMasterPort + ":" + redisClusterMasterPort, + redisClusterReplicaPort + ":" + redisClusterReplicaPort)); + redisCluster + .withExposedPorts(redisClusterMasterPort, redisClusterReplicaPort) + .withCommand( + "sh", "-c", redisClusterCommand(redisClusterMasterPort, redisClusterReplicaPort)) + .waitingFor(Wait.forLogMessage(".*CLUSTER_READY.*\\n", 1)); + redisCluster.start(); + + RedisURI redisURI = + RedisURI.Builder.redis(redisCluster.getHost(), redisClusterMasterPort).build(); + redisClient = RedisClusterClient.create(redisURI); + 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.close(); + } + } + + @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(); + } + + @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)); + connection.setReadFrom(ReadFrom.SLAVE); + new PollingConditions(30) + .delay(0.5) + .eventually( + () -> { + redisClient.reloadPartitions(); + 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() { + 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 = findCommandSpans(writer, "SET"); + 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)); + } + } + + 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"; + } +} 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")