complete async unit test and changes to implementation
using handleAsync with BiFunction because lettuce futures almost never throw exceptions on futures
improvement changes to come
ADD
dd-java-agent/instrumentation/lettuce-5/src/main/java/datadog/trace/instrumentation/lettuce/RedisAsyncBiFunction.java
+package datadog.trace.instrumentation.lettuce;
+
+import io.opentracing.Span;
+import io.opentracing.tag.Tags;
+import java.util.Collections;
+import java.util.function.BiFunction;
+
+/**
+ * Callback class to close the span on an error or a success in the RedisFuture returned by
+ * the lettuce async API
+ * @param <T> the normal completion result
+ * @param <U> the error
+ * @param <R> the return type, should be null since nothing else should happen from tracing standpoint
+ * after the span is closed
+ */
+public class RedisAsyncBiFunction<T extends Object, U extends Throwable, R extends Object>
+ implements BiFunction<T, Throwable, R> {
+
+ private final Span span;
+
+ public RedisAsyncBiFunction(Span span) {
+ this.span = span;
+ }
+
+ @Override
+ public R apply(T t, Throwable throwable) {
+ if (throwable != null) {
+ Tags.ERROR.set(this.span, true);
+ this.span.log(Collections.singletonMap("error.object", throwable));
+ }
+ this.span.finish();
+ return null;
+ }
+}

MODIFY
dd-java-agent/instrumentation/lettuce-5/src/main/java/datadog/trace/instrumentation/lettuce/RedisAsyncCommandsInstrumentation.java
dd-java-agent/instrumentation/lettuce-5/src/main/java/datadog/trace/instrumentation/lettuce/RedisAsyncCommandsInstrumentation.java
@@ -15,18 +15,14 @@ import io.opentracing.Span;
import io.opentracing.tag.Tags;
import io.opentracing.util.GlobalTracer;
import java.util.Collections;
-import java.util.function.Consumer;
-import java.util.function.Function;
import net.bytebuddy.agent.builder.AgentBuilder;
import net.bytebuddy.asm.Advice;
@AutoService(Instrumenter.class)
public class RedisAsyncCommandsInstrumentation extends Instrumenter.Configurable {
- private static final HelperInjector REDIS_ASYNC_COMMANDS_HELPERS =
- new HelperInjector(
- RedisAsyncCommandsInstrumentation.class.getName() + "$RedisAsyncConsumer",
- RedisAsyncCommandsInstrumentation.class.getName() + "$RedisAsyncExceptionFunction");
+ private static final HelperInjector REDIS_ASYNC_HELPERS =
+ new HelperInjector(RedisAsyncCommandsInstrumentation.class.getPackage().getName() + ".RedisAsyncBiFunction");
private static final String SERVICE_NAME = "redis";
private static final String COMPONENT_NAME = SERVICE_NAME + "-client";
@@ -34,11 +30,16 @@ public class RedisAsyncCommandsInstrumentation extends Instrumenter.Configurable
super(SERVICE_NAME);
}
+ @Override
+ protected boolean defaultEnabled() {
+ return false;
+ }
+
@Override
protected AgentBuilder apply(AgentBuilder agentBuilder) {
return agentBuilder
.type(named("io.lettuce.core.AbstractRedisAsyncCommands"))
- .transform(REDIS_ASYNC_COMMANDS_HELPERS)
+ .transform(REDIS_ASYNC_HELPERS)
.transform(DDTransformers.defaultTransformers())
.transform(
DDAdvice.create()
@@ -96,56 +97,9 @@ public class RedisAsyncCommandsInstrumentation extends Instrumenter.Configurable
return;
}
- // use lettuce's callback interface to ensure no logic is lost
- asyncCommand.onComplete(new RedisAsyncConsumer<>(scope.span()));
- // exception callbacks are not part of lettuce's api, so use CompletableFuture's, in the future if
- // lettuce does add some sort of exception callback handler, we should use that instead.
- asyncCommand.exceptionally(new RedisAsyncExceptionFunction<>(scope.span()));
+ // close spans on error or normal completion
+ asyncCommand.handleAsync(new RedisAsyncBiFunction<>(scope.span()));
scope.close();
}
}
-
- /**
- * Callback class to close the span on an error in the RedisFuture returned by the lettuce async
- * API
- *
- * @param <Throwable> the error
- * @param <R> the object to return as a result of this callback, should be null as we are only
- * closing the span and logging the error
- */
- public static class RedisAsyncExceptionFunction<Throwable, R extends Object>
- implements Function<Throwable, R> {
- private final Span span;
-
- public RedisAsyncExceptionFunction(Span span) {
- this.span = span;
- }
-
- @Override
- public R apply(Throwable throwable) {
- Tags.ERROR.set(this.span, true);
- this.span.log(Collections.singletonMap("error.object", throwable));
- this.span.finish();
- return null;
- }
- }
-
- /**
- * Callback class to close the span on a normal execution of the RedisFuture returned by the
- * lettice async API
- *
- * @param <T> the object (async result) from lettuce async API computation
- */
- public static class RedisAsyncConsumer<T> implements Consumer<T> {
- private final Span span;
-
- public RedisAsyncConsumer(Span span) {
- this.span = span;
- }
-
- @Override
- public void accept(Object o) {
- this.span.finish();
- }
- }
}

MODIFY
dd-java-agent/instrumentation/lettuce-5/src/main/java/datadog/trace/instrumentation/lettuce/RedisClientInstrumentation.java
dd-java-agent/instrumentation/lettuce-5/src/main/java/datadog/trace/instrumentation/lettuce/RedisClientInstrumentation.java
@@ -5,9 +5,10 @@ import static net.bytebuddy.matcher.ElementMatchers.*;
import com.google.auto.service.AutoService;
import datadog.trace.agent.tooling.DDAdvice;
import datadog.trace.agent.tooling.DDTransformers;
+import datadog.trace.agent.tooling.HelperInjector;
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.api.DDTags;
-import io.lettuce.core.RedisURI;
+import io.lettuce.core.*;
import io.opentracing.Scope;
import io.opentracing.Span;
import io.opentracing.tag.Tags;
@@ -19,16 +20,24 @@ import net.bytebuddy.asm.Advice;
@AutoService(Instrumenter.class)
public final class RedisClientInstrumentation extends Instrumenter.Configurable {
+ private static final HelperInjector REDIS_ASYNC_HELPERS =
+ new HelperInjector(RedisAsyncCommandsInstrumentation.class.getPackage().getName() + ".RedisAsyncBiFunction");
+
public static final String SERVICE_NAME = "redis";
public static final String COMPONENT_NAME = SERVICE_NAME + "-client";
public static final String RESOURCE_NAME_PREFIX = "CONNECT:";
- public static final String REDIS_URL_TAGNAME = "db.redis.url";
+ public static final String REDIS_URL_TAG_NAME = "db.redis.url";
public static final String REDIS_DB_INDEX_TAG_NAME = "db.redis.dbIndex";
public RedisClientInstrumentation() {
super(SERVICE_NAME);
}
+ @Override
+ protected boolean defaultEnabled() {
+ return false;
+ }
+
@Override
public AgentBuilder apply(final AgentBuilder agentBuilder) {
return agentBuilder
@@ -36,24 +45,25 @@ public final class RedisClientInstrumentation extends Instrumenter.Configurable
named("io.lettuce.core.RedisClient")
.and(hasSuperType(named("io.lettuce.core.AbstractRedisClient"))))
.transform(DDTransformers.defaultTransformers())
+ .transform(REDIS_ASYNC_HELPERS)
.transform(
DDAdvice.create()
.advice(
isMethod()
.and(isPrivate())
- .and(returns(hasSuperType(named("io.lettuce.core.api.StatefulConnection"))))
+ .and(returns(named("io.lettuce.core.ConnectionFuture")))
.and(nameStartsWith("connect"))
+ .and(nameEndsWith("Async"))
.and(takesArgument(1, named("io.lettuce.core.RedisURI"))),
- StatefulRedisConnectionAdvice.class.getName()))
+ ConnectionFutureAdvice.class.getName()))
.asDecorator();
}
- public static class StatefulRedisConnectionAdvice {
+ public static class ConnectionFutureAdvice {
@Advice.OnMethodEnter(suppress = Throwable.class)
public static Scope startSpan(@Advice.Argument(1) final RedisURI redisURI) {
-
- final Scope scope = GlobalTracer.get().buildSpan(SERVICE_NAME + ".query").startActive(true);
+ final Scope scope = GlobalTracer.get().buildSpan(SERVICE_NAME + ".query").startActive(false);
final Span span = scope.span();
Tags.DB_TYPE.set(span, SERVICE_NAME);
@@ -62,7 +72,7 @@ public final class RedisClientInstrumentation extends Instrumenter.Configurable
final String url =
redisURI.getHost() + ":" + redisURI.getPort() + "/" + redisURI.getDatabase();
- span.setTag(REDIS_URL_TAGNAME, url);
+ span.setTag(REDIS_URL_TAG_NAME, url);
span.setTag(REDIS_DB_INDEX_TAG_NAME, redisURI.getDatabase());
span.setTag(DDTags.RESOURCE_NAME, RESOURCE_NAME_PREFIX + url);
span.setTag(DDTags.SERVICE_NAME, SERVICE_NAME);
@@ -73,12 +83,19 @@ public final class RedisClientInstrumentation extends Instrumenter.Configurable
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
public static void stopSpan(
- @Advice.Enter final Scope scope, @Advice.Thrown final Throwable throwable) {
+ @Advice.Enter final Scope scope,
+ @Advice.Thrown final Throwable throwable,
+ @Advice.Return(readOnly = false) final ConnectionFuture connectionFuture) {
if (throwable != null) {
final Span span = scope.span();
Tags.ERROR.set(span, true);
span.log(Collections.singletonMap("error.object", throwable));
+ scope.close();
+ return;
}
+
+ // close spans on error or normal completion
+ connectionFuture.handleAsync(new RedisAsyncBiFunction<>(scope.span()));
scope.close();
}
}

ADD
dd-java-agent/instrumentation/lettuce-5/src/test/groovy/LettuceAsyncClientTest.groovy
+import datadog.trace.agent.test.AgentTestRunner
+import datadog.trace.instrumentation.lettuce.RedisAsyncCommandsInstrumentation
+import datadog.trace.instrumentation.lettuce.RedisClientInstrumentation
+import io.lettuce.core.ConnectionFuture
+import io.lettuce.core.RedisClient
+import io.lettuce.core.RedisFuture
+import io.lettuce.core.RedisURI
+import io.lettuce.core.api.StatefulConnection
+import io.lettuce.core.api.async.RedisAsyncCommands
+import io.lettuce.core.codec.StringCodec
+import redis.embedded.RedisServer
+import spock.lang.Shared
+import spock.util.concurrent.AsyncConditions
+
+import java.util.concurrent.TimeUnit
+import java.util.function.BiConsumer
+import java.util.function.BiFunction
+import java.util.function.Consumer
+import java.util.function.Function
+
+class LettuceAsyncClientTest extends AgentTestRunner {
+
+ static {
+ System.setProperty("dd.integration.redis.enabled", "true")
+ }
+
+ @Shared
+ public static final String HOST = "127.0.0.1"
+ public static final int PORT = 6399
+ public static final int INCORRECT_PORT = 9999
+ public static final int DB_INDEX = 0
+ @Shared
+ public static final String DB_ADDR = HOST + ":" + PORT + "/" + DB_INDEX
+ @Shared
+ public static final String DB_ADDR_NON_EXISTENT = HOST + ":" + INCORRECT_PORT + "/" + DB_INDEX
+ @Shared
+ public static final String DB_URI_NON_EXISTENT = "redis://" + DB_ADDR_NON_EXISTENT
+ public static final String EMBEDDED_DB_URI = "redis://" + DB_ADDR
+
+ @Shared
+ RedisServer redisServer = RedisServer.builder()
+ // bind to localhost to avoid firewall popup
+ .setting("bind " + HOST)
+ // set max memory to avoid problems in CI
+ .setting("maxmemory 128M")
+ .port(PORT).build()
+
+ @Shared
+ RedisClient redisClient = RedisClient.create(EMBEDDED_DB_URI)
+
+ @Shared
+ RedisAsyncCommands<String, ?> asyncCommands = null
+
+ @Shared
+ Map<String, String> testHashMap = [
+ firstname: "John",
+ lastname: "Doe",
+ age: "53"
+ ]
+
+ def setupSpec() {
+ println "Using redis: $redisServer.args"
+ redisServer.start()
+ StatefulConnection connection = redisClient.connect()
+ asyncCommands = connection.async()
+ }
+
+ def cleanupSpec() {
+ redisServer.stop()
+ }
+
+ def setup() {
+ TEST_WRITER.start()
+ }
+
+ def "connect using get on ConnectionFuture"() {
+ setup:
+ RedisClient testConnectionClient = RedisClient.create(EMBEDDED_DB_URI)
+ ConnectionFuture connectionFuture = testConnectionClient.connectAsync(StringCodec.UTF8,
+ new RedisURI(HOST, PORT, 3, TimeUnit.SECONDS))
+ def connection = connectionFuture.get()
+ TEST_WRITER.waitForTraces(1)
+
+ expect:
+ connection != null
+ TEST_WRITER.size() == 1
+
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+ def span = trace[0]
+ span.getServiceName() == RedisClientInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisClientInstrumentation.SERVICE_NAME
+ span.getResourceName() == RedisClientInstrumentation.RESOURCE_NAME_PREFIX + DB_ADDR
+ !span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags[RedisClientInstrumentation.REDIS_URL_TAG_NAME] == DB_ADDR
+ tags[RedisClientInstrumentation.REDIS_DB_INDEX_TAG_NAME] == 0
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisClientInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisClientInstrumentation.SERVICE_NAME
+ tags["component"] == RedisClientInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+
+ def "connect exception inside the connection future"() {
+ setup:
+ RedisClient testConnectionClient = RedisClient.create(DB_URI_NON_EXISTENT)
+ StatefulConnection connection = null
+ try {
+ ConnectionFuture connectionFuture = testConnectionClient.connectAsync(StringCodec.UTF8,
+ new RedisURI(HOST, INCORRECT_PORT, 3, TimeUnit.SECONDS))
+ connection = connectionFuture.get()
+ } catch (Exception rce) {
+ // do nothing, this is expected
+ println("caught " + rce.getMessage())
+ }
+
+ expect:
+ TEST_WRITER.waitForTraces(1)
+ connection == null
+ TEST_WRITER.size() == 1
+
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+ def span = trace[0]
+ span.getServiceName() == RedisClientInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisClientInstrumentation.SERVICE_NAME
+ span.getResourceName() == RedisClientInstrumentation.RESOURCE_NAME_PREFIX + DB_ADDR_NON_EXISTENT
+ span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags[RedisClientInstrumentation.REDIS_URL_TAG_NAME] == DB_ADDR_NON_EXISTENT
+ tags[RedisClientInstrumentation.REDIS_DB_INDEX_TAG_NAME] == 0
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisClientInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisClientInstrumentation.SERVICE_NAME
+ tags["component"] == RedisClientInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+
+ def "set command using Future get with timeout"() {
+ setup:
+ RedisFuture<String> redisFuture = asyncCommands.set("TESTKEY", "TESTVAL")
+ String res = redisFuture.get(3, TimeUnit.SECONDS)
+ TEST_WRITER.waitForTraces(1)
+
+ expect:
+ res == "OK"
+ TEST_WRITER.size() == 1
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+ def span = trace[0]
+ span.getServiceName() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getResourceName() == "SET"
+ !span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.redis.command.args"] == "key<TESTKEY> value<TESTVAL>"
+ tags["component"] == RedisAsyncCommandsInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+
+ def "get command chained with thenAccept"() {
+ setup:
+ def conds = new AsyncConditions()
+ Consumer<String> consumer = new Consumer<String>() {
+ @Override
+ void accept(String res) {
+ conds.evaluate {
+ assert res == "TESTVAL"
+ }
+ }
+ }
+
+ when:
+ RedisFuture<String> redisFuture = asyncCommands.get("TESTKEY")
+ redisFuture.thenAccept(consumer)
+
+ then:
+ conds.await()
+ TEST_WRITER.waitForTraces(1)
+ TEST_WRITER.size() == 1
+
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+ def span = trace[0]
+ span.getServiceName() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getResourceName() == "GET"
+ !span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.redis.command.args"] == "key<TESTKEY>"
+ tags["component"] == RedisAsyncCommandsInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+
+ // to make sure instrumentation's chained completion stages won't interfere with user's, while still
+ // recording metrics
+ def "get non existent key command with handleAsync and chained with thenApply"() {
+ setup:
+ def conds = new AsyncConditions()
+ final String SUCCESS = "KEY MISSING"
+ BiFunction<String, Throwable, String> firstStage = new BiFunction<String, Throwable, String>() {
+ @Override
+ String apply(String res, Throwable throwable) {
+ conds.evaluate{
+ assert res == null
+ assert throwable == null
+ }
+ return (res == null ? SUCCESS : res)
+ }
+ }
+ Function<String, Object> secondStage = new Function<String, Object>() {
+ @Override
+ Object apply(String input) {
+ conds.evaluate{
+ assert input == SUCCESS
+ }
+ return null
+ }
+ }
+
+ when:
+ RedisFuture<String> redisFuture = asyncCommands.get("NON_EXISTENT_KEY")
+ redisFuture.handleAsync(firstStage).thenApply(secondStage)
+
+ then:
+ conds.await()
+ TEST_WRITER.waitForTraces(1)
+ TEST_WRITER.size() == 1
+
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+ def span = trace[0]
+ span.getServiceName() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getResourceName() == "GET"
+ !span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.redis.command.args"] == "key<NON_EXISTENT_KEY>"
+ tags["component"] == RedisAsyncCommandsInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+
+ def "command with no arguments using a biconsumer"() {
+ setup:
+ def conds = new AsyncConditions()
+ BiConsumer<String, Throwable> biConsumer = new BiConsumer<String, Throwable>() {
+ @Override
+ void accept(String keyRetrieved, Throwable throwable) {
+ conds.evaluate{
+ assert keyRetrieved == "TESTKEY"
+ }
+ }
+ }
+
+ when:
+ RedisFuture<String> redisFuture = asyncCommands.randomkey()
+ redisFuture.whenCompleteAsync(biConsumer)
+
+ then:
+ conds.await()
+ TEST_WRITER.waitForTraces(1)
+ TEST_WRITER.size() == 1
+
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+
+ def span = trace[0]
+ span.getServiceName() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getResourceName() == "RANDOMKEY"
+ !span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.redis.command.args"] == null
+ tags["component"] == RedisAsyncCommandsInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+
+ def "hash set and then nest apply to hash getall"() {
+ setup:
+ def conds = new AsyncConditions()
+
+ when:
+ RedisFuture<String> hmsetFuture = asyncCommands.hmset("user", testHashMap)
+ hmsetFuture.thenApplyAsync(new Function<String, Object>() {
+ @Override
+ Object apply(String setResult) {
+ TEST_WRITER.waitForTraces(1)
+ TEST_WRITER.start()
+ conds.evaluate {
+ assert setResult == "OK"
+ }
+ RedisFuture<Map<String, String>> hmGetAllFuture = asyncCommands.hgetall("user")
+ hmGetAllFuture.exceptionally(new Function<Throwable, Map<String, String>>() {
+ @Override
+ Map<String, String> apply(Throwable throwable) {
+ println("unexpected:" + throwable.toString())
+ throwable.printStackTrace()
+ assert false
+ return null
+ }
+ })
+ hmGetAllFuture.thenAccept(new Consumer<Map<String, String>>() {
+ @Override
+ void accept(Map<String, String> hmGetAllResult) {
+ conds.evaluate {
+ assert testHashMap == hmGetAllResult
+ }
+ }
+ })
+ return null
+ }
+ })
+
+ then:
+ conds.await()
+ TEST_WRITER.waitForTraces(1)
+ def trace = TEST_WRITER.firstTrace()
+ trace.size() == 1
+
+ def span = trace[0]
+ span.getServiceName() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getOperationName() == "redis.query"
+ span.getType() == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ span.getResourceName() == "HGETALL"
+ !span.context().getErrorFlag()
+
+ def tags = span.context().tags
+ tags["span.kind"] == "client"
+ tags["span.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.type"] == RedisAsyncCommandsInstrumentation.SERVICE_NAME
+ tags["db.redis.command.args"] == "key<user>"
+ tags["component"] == RedisAsyncCommandsInstrumentation.COMPONENT_NAME
+ tags["thread.name"] != null
+ tags["thread.id"] != null
+ }
+}

MODIFY
dd-java-agent/instrumentation/lettuce-5/src/test/groovy/LettuceSyncClientTest.groovy
dd-java-agent/instrumentation/lettuce-5/src/test/groovy/LettuceSyncClientTest.groovy
@@ -10,6 +10,10 @@ import spock.lang.Shared
class LettuceSyncClientTest extends AgentTestRunner {
+ static {
+ System.setProperty("dd.integration.redis.enabled", "true")
+ }
+
@Shared
public static final String HOST = "127.0.0.1"
public static final int PORT = 6399
@@ -62,6 +66,7 @@ class LettuceSyncClientTest extends AgentTestRunner {
setup:
RedisClient testConnectionClient = RedisClient.create(EMBEDDED_DB_URI)
testConnectionClient.connect()
+ TEST_WRITER.waitForTraces(1)
expect:
TEST_WRITER.size() == 1
@@ -76,7 +81,7 @@ class LettuceSyncClientTest extends AgentTestRunner {
!span.context().getErrorFlag()
def tags = span.context().tags
- tags[RedisClientInstrumentation.REDIS_URL_TAGNAME] == DB_ADDR
+ tags[RedisClientInstrumentation.REDIS_URL_TAG_NAME] == DB_ADDR
tags[RedisClientInstrumentation.REDIS_DB_INDEX_TAG_NAME] == 0
tags["span.kind"] == "client"
tags["span.type"] == RedisClientInstrumentation.SERVICE_NAME
@@ -92,6 +97,7 @@ class LettuceSyncClientTest extends AgentTestRunner {
try {
testConnectionClient.connect()
} catch (RedisConnectionException rce) { }
+ TEST_WRITER.waitForTraces(1)
expect:
TEST_WRITER.size() == 1
@@ -106,7 +112,7 @@ class LettuceSyncClientTest extends AgentTestRunner {
span.context().getErrorFlag()
def tags = span.context().tags
- tags[RedisClientInstrumentation.REDIS_URL_TAGNAME] == DB_ADDR_NON_EXISTENT
+ tags[RedisClientInstrumentation.REDIS_URL_TAG_NAME] == DB_ADDR_NON_EXISTENT
tags[RedisClientInstrumentation.REDIS_DB_INDEX_TAG_NAME] == 0
tags["span.kind"] == "client"
tags["span.type"] == RedisClientInstrumentation.SERVICE_NAME

complete async unit test and changes to implementation

using handleAsync with BiFunction because lettuce futures almost never throw exceptions on futures

improvement changes to come