diff --git a/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedisCommandDataInstrumentation.java b/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedisCommandDataInstrumentation.java index 2d974b1da8e8..deb513424f07 100644 --- a/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedisCommandDataInstrumentation.java +++ b/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedisCommandDataInstrumentation.java @@ -13,6 +13,7 @@ import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; import io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.CompletableFutureWrapper; +import io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.RedissonErrorCapture; import java.util.concurrent.CompletableFuture; import net.bytebuddy.asm.Advice; import net.bytebuddy.asm.Advice.AssignReturned; @@ -26,7 +27,9 @@ class RedisCommandDataInstrumentation implements TypeInstrumentation { @Override public ElementMatcher typeMatcher() { return namedOneOf( - "org.redisson.client.protocol.CommandData", "org.redisson.client.protocol.CommandsData"); + "org.redisson.client.protocol.CommandData", + "org.redisson.client.protocol.CommandsData", + "org.redisson.client.protocol.BatchCommandData"); } @Override @@ -39,6 +42,9 @@ public void transform(TypeTransformer transformer) { transformer.applyAdviceToMethod( isConstructor().and(takesArgument(0, CompletableFuture.class)), getClass().getName() + "$WrapCompletableFutureAdvice"); + transformer.applyAdviceToMethod( + named("tryFailure").and(takesArgument(0, Throwable.class)), + getClass().getName() + "$CaptureErrorAdvice"); } @SuppressWarnings("unused") @@ -61,4 +67,13 @@ public static CompletableFuture onEnter( return CompletableFutureWrapper.wrap(completableFuture); } } + + @SuppressWarnings("unused") + public static class CaptureErrorAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class, inline = false) + public static void onEnter(@Advice.Argument(0) Throwable error) { + RedissonErrorCapture.captureServerError(error); + } + } } diff --git a/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonInstrumentationModule.java b/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonInstrumentationModule.java index 6acbe22829e6..171c13ab2b5d 100644 --- a/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonInstrumentationModule.java +++ b/instrumentation/redisson/redisson-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonInstrumentationModule.java @@ -13,6 +13,7 @@ import com.google.auto.service.AutoService; import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule; import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; +import io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.RedisCommandDecoderInstrumentation; import java.util.List; import net.bytebuddy.matcher.ElementMatcher; @@ -35,6 +36,7 @@ public List typeInstrumentations() { new ConnectionManagerConnectionFutureInstrumentation(), new MasterSlaveConnectionManagerInstrumentation(), new RedisConnectionInstrumentation(), + new RedisCommandDecoderInstrumentation(), new RedisCommandDataInstrumentation()); } diff --git a/instrumentation/redisson/redisson-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonAsyncClientTest.java b/instrumentation/redisson/redisson-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonAsyncClientTest.java index 19412ef7b152..a647bc576645 100644 --- a/instrumentation/redisson/redisson-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonAsyncClientTest.java +++ b/instrumentation/redisson/redisson-3.0/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_0/RedissonAsyncClientTest.java @@ -6,5 +6,12 @@ package io.opentelemetry.javaagent.instrumentation.redisson.v3_0; import io.opentelemetry.javaagent.instrumentation.redisson.AbstractRedissonAsyncClientTest; +import org.redisson.api.RBatch; +import org.redisson.api.RedissonClient; -class RedissonAsyncClientTest extends AbstractRedissonAsyncClientTest {} +class RedissonAsyncClientTest extends AbstractRedissonAsyncClientTest { + @Override + protected RBatch createBatch(RedissonClient redisson) { + return redisson.createBatch(); + } +} diff --git a/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedisCommandDataInstrumentation.java b/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedisCommandDataInstrumentation.java index 46ffda2f4853..e54b955a85b6 100644 --- a/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedisCommandDataInstrumentation.java +++ b/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedisCommandDataInstrumentation.java @@ -6,12 +6,14 @@ package io.opentelemetry.javaagent.instrumentation.redisson.v3_17; import static net.bytebuddy.matcher.ElementMatchers.isConstructor; +import static net.bytebuddy.matcher.ElementMatchers.named; import static net.bytebuddy.matcher.ElementMatchers.namedOneOf; import static net.bytebuddy.matcher.ElementMatchers.takesArgument; import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; import io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.CompletableFutureWrapper; +import io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.RedissonErrorCapture; import java.util.concurrent.CompletableFuture; import net.bytebuddy.asm.Advice; import net.bytebuddy.asm.Advice.AssignReturned; @@ -24,7 +26,9 @@ class RedisCommandDataInstrumentation implements TypeInstrumentation { @Override public ElementMatcher typeMatcher() { return namedOneOf( - "org.redisson.client.protocol.CommandData", "org.redisson.client.protocol.CommandsData"); + "org.redisson.client.protocol.CommandData", + "org.redisson.client.protocol.CommandsData", + "org.redisson.client.protocol.BatchCommandData"); } @Override @@ -32,6 +36,9 @@ public void transform(TypeTransformer transformer) { transformer.applyAdviceToMethod( isConstructor().and(takesArgument(0, CompletableFuture.class)), getClass().getName() + "$WrapCompletableFutureAdvice"); + transformer.applyAdviceToMethod( + named("tryFailure").and(takesArgument(0, Throwable.class)), + getClass().getName() + "$CaptureErrorAdvice"); } @SuppressWarnings("unused") @@ -44,4 +51,13 @@ public static CompletableFuture onEnter( return CompletableFutureWrapper.wrap(completableFuture); } } + + @SuppressWarnings("unused") + public static class CaptureErrorAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class, inline = false) + public static void onEnter(@Advice.Argument(0) Throwable error) { + RedissonErrorCapture.captureServerError(error); + } + } } diff --git a/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedissonInstrumentationModule.java b/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedissonInstrumentationModule.java index 6ca9070d4c33..46e76c5b4738 100644 --- a/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedissonInstrumentationModule.java +++ b/instrumentation/redisson/redisson-3.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/v3_17/RedissonInstrumentationModule.java @@ -12,6 +12,7 @@ import com.google.auto.service.AutoService; import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule; import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; +import io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.RedisCommandDecoderInstrumentation; import java.util.List; import net.bytebuddy.matcher.ElementMatcher; @@ -34,6 +35,7 @@ public List typeInstrumentations() { new MasterSlaveConnectionManagerInstrumentation(), new RedisExecutorConnectionFutureInstrumentation(), new RedisConnectionInstrumentation(), + new RedisCommandDecoderInstrumentation(), new RedisCommandDataInstrumentation()); } diff --git a/instrumentation/redisson/redisson-common-3.0/javaagent/build.gradle.kts b/instrumentation/redisson/redisson-common-3.0/javaagent/build.gradle.kts index 3328b21686c0..744f2d586adb 100644 --- a/instrumentation/redisson/redisson-common-3.0/javaagent/build.gradle.kts +++ b/instrumentation/redisson/redisson-common-3.0/javaagent/build.gradle.kts @@ -8,3 +8,21 @@ dependencies { compileOnly("com.google.auto.value:auto-value-annotations") annotationProcessor("com.google.auto.value:auto-value") } + +testing { + suites { + register("unitTests") { + dependencies { + implementation(project()) + implementation(project(":instrumentation-api-incubator")) + implementation("org.redisson:redisson:3.0.0") + } + } + } +} + +tasks { + check { + dependsOn(testing.suites) + } +} diff --git a/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/EndOperationListener.java b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/EndOperationListener.java index 6d252a995098..937b1374e3d4 100644 --- a/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/EndOperationListener.java +++ b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/EndOperationListener.java @@ -24,6 +24,10 @@ public EndOperationListener( @Override public void accept(@Nullable T unused, @Nullable Throwable error) { - instrumenter.end(context, request, null, error); + try { + instrumenter.end(context, request, null, error); + } finally { + RedissonErrorCapture.clearServerError(error); + } } } diff --git a/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedisCommandDecoderInstrumentation.java b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedisCommandDecoderInstrumentation.java new file mode 100644 index 000000000000..40b865acd111 --- /dev/null +++ b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedisCommandDecoderInstrumentation.java @@ -0,0 +1,50 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0; + +import static io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.RedissonErrorCapture.currentServerErrorType; +import static io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0.RedissonErrorCapture.parseServerErrorType; +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; + +import io.netty.buffer.ByteBuf; +import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; +import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; +import javax.annotation.Nullable; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; + +public class RedisCommandDecoderInstrumentation implements TypeInstrumentation { + + @Override + public ElementMatcher typeMatcher() { + return named("org.redisson.client.handler.CommandDecoder"); + } + + @Override + public void transform(TypeTransformer transformer) { + transformer.applyAdviceToMethod( + named("decode") + .and(takesArgument(0, named("io.netty.buffer.ByteBuf"))) + .and(takesArgument(1, named("org.redisson.client.protocol.CommandData"))), + getClass().getName() + "$DecodeAdvice"); + } + + @SuppressWarnings("unused") + public static class DecodeAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class, inline = false) + public static String onEnter(@Advice.Argument(0) ByteBuf input) { + return currentServerErrorType().set(parseServerErrorType(input)); + } + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class, inline = false) + public static void onExit(@Advice.Enter @Nullable String previousErrorType) { + currentServerErrorType().restore(previousErrorType); + } + } +} diff --git a/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetter.java b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetter.java index 1bbe40e94c4d..7493376cfc30 100644 --- a/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetter.java +++ b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetter.java @@ -43,6 +43,13 @@ public Long getDbOperationBatchSize(RedissonRequest request) { return request.getOperationBatchSize(); } + @Override + @Nullable + public String getErrorType( + RedissonRequest request, @Nullable Void response, @Nullable Throwable error) { + return RedissonErrorCapture.getServerErrorType(error); + } + @Nullable @Override public String getServerAddress(RedissonRequest request) { diff --git a/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonErrorCapture.java b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonErrorCapture.java new file mode 100644 index 000000000000..cf9c19bdbb67 --- /dev/null +++ b/instrumentation/redisson/redisson-common-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonErrorCapture.java @@ -0,0 +1,103 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import io.netty.buffer.ByteBuf; +import io.opentelemetry.instrumentation.api.incubator.semconv.db.internal.RedisErrorPrefix; +import io.opentelemetry.instrumentation.api.internal.ScopedThreadValue; +import io.opentelemetry.instrumentation.api.util.VirtualField; +import javax.annotation.Nullable; +import org.redisson.client.RedisException; + +/** + * Captures the first token from RESP error replies before Redisson completes the command promise. + * + *

Redisson's {@code CommandDecoder} calls {@code CommandData.tryFailure} while decoding a {@code + * -} reply, so the token is scoped to that decoder frame and attached to the resulting Redisson + * exception before completion callbacks can run. The operation listener clears it after extracting + * telemetry; all other errors use the shared exception-class fallback. + * + * @see RESP + * specification + * @see Redisson + * CommandDecoder + */ +public final class RedissonErrorCapture { + + private static final ScopedThreadValue currentServerErrorType = new ScopedThreadValue<>(); + + private static final VirtualField SERVER_ERROR = + VirtualField.find(RedisException.class, String.class, "redissonServerErrorType"); + + public static ScopedThreadValue currentServerErrorType() { + return currentServerErrorType; + } + + @Nullable + public static String parseServerErrorType(ByteBuf input) { + try { + int replyStart = input.readerIndex(); + if (input.getByte(replyStart) != '-') { + return null; + } + + int lineStart = replyStart + 1; + int lineEnd = findCrlf(input, lineStart); + if (lineEnd == -1) { + return null; + } + + return RedisErrorPrefix.extract(input.toString(lineStart, lineEnd - lineStart, UTF_8)); + } catch (Throwable ignored) { + // ReplayingDecoder retries incomplete replies; leave buffer handling to Redisson. + return null; + } + } + + private static int findCrlf(ByteBuf input, int start) { + int end = input.writerIndex(); + for (int i = start; i + 1 < end; i++) { + if (input.getByte(i) == '\r' && input.getByte(i + 1) == '\n') { + return i; + } + } + return -1; + } + + public static void captureServerError(@Nullable Throwable error) { + if (!(error instanceof RedisException)) { + return; + } + + String errorType = currentServerErrorType.get(); + if (errorType != null) { + SERVER_ERROR.set((RedisException) error, errorType); + // consume so completion callbacks run by tryFailure cannot attribute it to another command + currentServerErrorType.set(null); + } + } + + @Nullable + public static String getServerErrorType(@Nullable Throwable error) { + if (!(error instanceof RedisException)) { + return null; + } + + return SERVER_ERROR.get((RedisException) error); + } + + public static void clearServerError(@Nullable Throwable error) { + if (error instanceof RedisException) { + SERVER_ERROR.set((RedisException) error, null); + } + } + + private RedissonErrorCapture() {} +} diff --git a/instrumentation/redisson/redisson-common-3.0/javaagent/src/unitTests/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetterTest.java b/instrumentation/redisson/redisson-common-3.0/javaagent/src/unitTests/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetterTest.java new file mode 100644 index 000000000000..d84a79d9c290 --- /dev/null +++ b/instrumentation/redisson/redisson-common-3.0/javaagent/src/unitTests/java/io/opentelemetry/javaagent/instrumentation/redisson/common/v3_0/RedissonDbAttributesGetterTest.java @@ -0,0 +1,180 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.redisson.common.v3_0; + +import static io.opentelemetry.semconv.ErrorAttributes.ERROR_TYPE; +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.context.Context; +import io.opentelemetry.instrumentation.api.incubator.semconv.db.DbClientAttributesExtractor; +import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor; +import org.junit.jupiter.api.Test; +import org.redisson.client.RedisConnectionException; +import org.redisson.client.RedisException; +import org.redisson.client.RedisLoadingException; + +class RedissonDbAttributesGetterTest { + + private static final RedissonDbAttributesGetter GETTER = new RedissonDbAttributesGetter(); + private static final RedissonRequest REQUEST = + RedissonRequest.create(null, new Object(), null, null); + private static final AttributesExtractor EXTRACTOR = + DbClientAttributesExtractor.create(GETTER); + + @Test + void capturesUnknownServerPrefixFromRespReply() { + ByteBuf response = Unpooled.copiedBuffer("-CUSTOM diagnostic details\r\n", US_ASCII); + String previous = + RedissonErrorCapture.currentServerErrorType() + .set(RedissonErrorCapture.parseServerErrorType(response)); + RedisException error = new RedisException("CUSTOM diagnostic details. channel: redis"); + try { + RedissonErrorCapture.captureServerError(error); + } finally { + RedissonErrorCapture.currentServerErrorType().restore(previous); + response.release(); + } + + assertThat(GETTER.getErrorType(REQUEST, null, error)).isEqualTo("CUSTOM"); + assertThat(extractedErrorType(error)).isEqualTo("CUSTOM"); + + RedissonErrorCapture.clearServerError(error); + assertThat(extractedErrorType(error)).isEqualTo(RedisException.class.getName()); + assertThat(RedissonErrorCapture.currentServerErrorType().get()).isNull(); + assertThat(extractedErrorType(new RedisException("CUSTOM other operation"))) + .isEqualTo(RedisException.class.getName()); + } + + @Test + void malformedAndNonErrorRepliesDoNotProduceServerErrorType() { + ByteBuf success = Unpooled.copiedBuffer("+OK\r\n", US_ASCII); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(success)).isNull(); + } finally { + success.release(); + } + + ByteBuf emptyPrefix = Unpooled.copiedBuffer("- \r\n", US_ASCII); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(emptyPrefix)).isNull(); + } finally { + emptyPrefix.release(); + } + + ByteBuf incomplete = Unpooled.copiedBuffer("-CUSTOM", US_ASCII); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(incomplete)).isNull(); + } finally { + incomplete.release(); + } + + ByteBuf incompleteAfterToken = Unpooled.copiedBuffer("-CUSTOM details", US_ASCII); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(incompleteAfterToken)).isNull(); + } finally { + incompleteAfterToken.release(); + } + + ByteBuf emptyReply = Unpooled.copiedBuffer("-\r\n", US_ASCII); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(emptyReply)).isNull(); + } finally { + emptyReply.release(); + } + + ByteBuf emptyToken = Unpooled.copiedBuffer("- \r\n", US_ASCII); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(emptyToken)).isNull(); + } finally { + emptyToken.release(); + } + } + + @Test + void parsingStopsAtTheReplyTerminatorWithoutConsumingTheBuffer() { + ByteBuf response = + Unpooled.copiedBuffer("ignored:-PréfixOnly\r\n-SECOND diagnostic\r\n", UTF_8); + response.readerIndex(8); + try { + assertThat(RedissonErrorCapture.parseServerErrorType(response)).isEqualTo("PréfixOnly"); + assertThat(response.readerIndex()).isEqualTo(8); + } finally { + response.release(); + } + } + + @Test + void clientExceptionMessageIsNotClassifiedAsServerReply() { + RedisException error = new RedisException("ERR connection refused"); + + RedissonErrorCapture.captureServerError(error); + + assertThat(GETTER.getErrorType(REQUEST, null, error)).isNull(); + assertThat(extractedErrorType(error)).isEqualTo(RedisException.class.getName()); + } + + @Test + void dedicatedExceptionUsesCapturedServerPrefixAndClearsIt() { + RedisLoadingException error = new RedisLoadingException("LOADING Redis is loading"); + ByteBuf response = Unpooled.copiedBuffer("-LOADING Redis is loading\r\n", US_ASCII); + String previous = + RedissonErrorCapture.currentServerErrorType() + .set(RedissonErrorCapture.parseServerErrorType(response)); + try { + RedissonErrorCapture.captureServerError(error); + } finally { + RedissonErrorCapture.currentServerErrorType().restore(previous); + response.release(); + } + + assertThat(GETTER.getErrorType(REQUEST, null, error)).isEqualTo("LOADING"); + assertThat(extractedErrorType(error)).isEqualTo("LOADING"); + + RedissonErrorCapture.clearServerError(error); + assertThat(extractedErrorType(error)).isEqualTo(RedisLoadingException.class.getName()); + assertThat(RedissonErrorCapture.currentServerErrorType().get()).isNull(); + } + + @Test + void capturedPrefixIsConsumedBeforeCompletionCallbacks() { + RedisException serverError = new RedisException("CUSTOM diagnostic details"); + RedisException clientError = new RedisException("client-side failure"); + String previous = RedissonErrorCapture.currentServerErrorType().set("CUSTOM"); + try { + RedissonErrorCapture.captureServerError(serverError); + // a callback issuing a new, client-side failing command within the decoder frame + RedissonErrorCapture.captureServerError(clientError); + assertThat(RedissonErrorCapture.currentServerErrorType().get()).isNull(); + // a repeated capture, as with BatchCommandData calling super.tryFailure, keeps the prefix + RedissonErrorCapture.captureServerError(serverError); + } finally { + RedissonErrorCapture.currentServerErrorType().restore(previous); + } + + assertThat(extractedErrorType(serverError)).isEqualTo("CUSTOM"); + assertThat(extractedErrorType(clientError)).isEqualTo(RedisException.class.getName()); + } + + @Test + void connectionExceptionIsNotClassifiedAsServerError() { + RedisConnectionException error = new RedisConnectionException("ERR connection refused"); + + assertThat(GETTER.getErrorType(REQUEST, null, error)).isNull(); + assertThat(extractedErrorType(error)).isEqualTo(RedisConnectionException.class.getName()); + } + + private static String extractedErrorType(Throwable error) { + AttributesBuilder attributes = Attributes.builder(); + EXTRACTOR.onEnd(attributes, Context.root(), REQUEST, null, error); + return attributes.build().get(ERROR_TYPE); + } +} diff --git a/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonAsyncClientTest.java b/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonAsyncClientTest.java index 9abf6eff74d9..058b982dc189 100644 --- a/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonAsyncClientTest.java +++ b/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonAsyncClientTest.java @@ -14,13 +14,16 @@ import static io.opentelemetry.semconv.DbAttributes.DB_OPERATION_NAME; import static io.opentelemetry.semconv.DbAttributes.DB_QUERY_TEXT; import static io.opentelemetry.semconv.DbAttributes.DB_SYSTEM_NAME; +import static io.opentelemetry.semconv.ErrorAttributes.ERROR_TYPE; import static io.opentelemetry.semconv.NetworkAttributes.NETWORK_PEER_ADDRESS; import static io.opentelemetry.semconv.NetworkAttributes.NETWORK_PEER_PORT; import static io.opentelemetry.semconv.ServerAttributes.SERVER_ADDRESS; import static io.opentelemetry.semconv.ServerAttributes.SERVER_PORT; import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DbSystemNameIncubatingValues.REDIS; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.awaitility.Awaitility.await; import io.opentelemetry.api.trace.Span; @@ -37,6 +40,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assumptions; @@ -54,9 +58,12 @@ import org.redisson.api.RFuture; import org.redisson.api.RScheduledExecutorService; import org.redisson.api.RScheduledFuture; +import org.redisson.api.RScript; import org.redisson.api.RSet; import org.redisson.api.RedissonClient; +import org.redisson.client.RedisException; import org.redisson.client.codec.StringCodec; +import org.redisson.client.protocol.Decoder; import org.redisson.config.Config; import org.redisson.config.SingleServerConfig; import org.testcontainers.containers.GenericContainer; @@ -103,6 +110,12 @@ void setup(TestInfo testInfo) throws InvocationTargetException, IllegalAccessExc newAddress = "redis://" + address; } Config config = new Config(); + try { + // script cache is enabled by default in 3.46.0 + Config.class.getMethod("setUseScriptCache", boolean.class).invoke(config, false); + } catch (NoSuchMethodException ignored) { + // ignored + } SingleServerConfig singleServerConfig = config.useSingleServer(); singleServerConfig.setAddress(newAddress); singleServerConfig.setTimeout(30_000); @@ -164,6 +177,113 @@ void futureSet() { equalTo(DB_OPERATION_NAME, "SET")))); } + @Test + void asyncCustomServerErrorIsRecordedInSpanAndMetrics() { + RFuture future = + redisson + .getScript() + .evalAsync( + RScript.Mode.READ_WRITE, + "return redis.error_reply('CUSTOM diagnostic details')", + RScript.ReturnType.VALUE); + assertThatThrownBy(() -> future.toCompletableFuture().get(TIMEOUT.toMillis(), MILLISECONDS)) + .hasCauseInstanceOf(RedisException.class); + + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(0).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> + span.hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()).containsEntry(ERROR_TYPE, "CUSTOM"))); + }); + + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram.hasPointsSatisfying( + point -> + point.hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, "CUSTOM"))))); + } + + @Test + void asyncWrongTypeErrorIsRecordedInSpanAndMetrics() throws IOException, InterruptedException { + assertThat( + redisServer + .execInContainer("redis-cli", "lpush", "async-wrong-type", "value") + .getExitCode()) + .isZero(); + assertThatThrownBy( + () -> + redisson + .getBucket("async-wrong-type") + .getAsync() + .toCompletableFuture() + .join()) + .hasRootCauseInstanceOf(RedisException.class); + assertErrorTypeOnSpanAndMetric("WRONGTYPE"); + + assertThat( + redisson + .getScript() + .evalAsync(RScript.Mode.READ_ONLY, "return 1", RScript.ReturnType.VALUE) + .toCompletableFuture() + .join()) + .isEqualTo(1L); + assertErrorTypeOnSpanAndMetric(null); + } + + @Test + void asyncFailedBatchErrorPrefixIsRecordedInSpanAndMetrics() { + RBatch batch = createBatch(redisson); + batch + .getScript() + .evalAsync( + RScript.Mode.READ_WRITE, + "return redis.error_reply('Batch.Code! diagnostic')", + RScript.ReturnType.VALUE); + + assertThatThrownBy(() -> batch.executeAsync().toCompletableFuture().join()) + .hasRootCauseInstanceOf(RedisException.class); + + assertErrorTypeOnSpanAndMetric("Batch.Code!"); + } + + @Test + void asyncClientSideDecodeFailureKeepsExceptionClassFallback() + throws IOException, InterruptedException { + assertThat( + redisServer + .execInContainer("redis-cli", "set", "async-client-side-decode-error", "value") + .getExitCode()) + .isZero(); + RBucket bucket = + redisson.getBucket( + "async-client-side-decode-error", + new StringCodec() { + @Override + public Decoder getValueDecoder() { + return (buffer, state) -> { + throw new RedisException("ERR client-side decode failure"); + }; + } + }); + assertThatThrownBy(() -> bucket.getAsync().toCompletableFuture().join()) + .hasRootCauseInstanceOf(RedisException.class); + + assertErrorTypeOnSpanAndMetric(RedisException.class.getName()); + } + @Test @Tag(TEST_RECONNECT) void futureCallback() { @@ -202,6 +322,48 @@ void futureCallback() { span -> span.hasName("callback").hasKind(INTERNAL).hasParent(trace.getSpan(0)))); } + private static void assertErrorTypeOnSpanAndMetric(String expectedErrorType) { + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(0).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> + span.hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> { + if (expectedErrorType == null) { + assertThat(attributes.asMap()).doesNotContainKey(ERROR_TYPE); + } else { + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, expectedErrorType); + } + })); + }); + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram.hasPointsSatisfying( + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> { + if (expectedErrorType == null) { + assertThat(attributes.asMap()) + .doesNotContainKey(ERROR_TYPE); + } else { + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, expectedErrorType); + } + })))); + testing.clearData(); + } + // regression test for // https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/19332 @Test @@ -362,6 +524,10 @@ private String dbNamespace(String databaseIndex) { return hasDatabaseIndex() ? databaseIndex : null; } + protected RBatch createBatch(RedissonClient redisson) { + return redisson.createBatch(BatchOptions.defaults()); + } + private static class MyCallable implements Serializable, Callable { private static final long serialVersionUID = 1L; diff --git a/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonClientTest.java b/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonClientTest.java index b61d0c1ae180..225ac4f0cfc8 100644 --- a/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonClientTest.java +++ b/instrumentation/redisson/redisson-common-3.0/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/redisson/AbstractRedissonClientTest.java @@ -19,6 +19,7 @@ import static io.opentelemetry.semconv.DbAttributes.DB_OPERATION_NAME; import static io.opentelemetry.semconv.DbAttributes.DB_QUERY_TEXT; import static io.opentelemetry.semconv.DbAttributes.DB_SYSTEM_NAME; +import static io.opentelemetry.semconv.ErrorAttributes.ERROR_TYPE; import static io.opentelemetry.semconv.NetworkAttributes.NETWORK_PEER_ADDRESS; import static io.opentelemetry.semconv.NetworkAttributes.NETWORK_PEER_PORT; import static io.opentelemetry.semconv.ServerAttributes.SERVER_ADDRESS; @@ -26,6 +27,8 @@ import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DbSystemNameIncubatingValues.REDIS; import static java.util.Arrays.asList; import static java.util.Collections.nCopies; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.jupiter.params.provider.Arguments.argumentSet; import io.opentelemetry.api.common.AttributeKey; @@ -33,13 +36,19 @@ import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; import io.opentelemetry.sdk.testing.assertj.TraceAssert; +import java.io.IOException; import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.stream.Stream; @@ -61,12 +70,18 @@ import org.redisson.api.RAtomicLong; import org.redisson.api.RBatch; import org.redisson.api.RBucket; +import org.redisson.api.RFuture; import org.redisson.api.RList; import org.redisson.api.RLock; import org.redisson.api.RMap; import org.redisson.api.RScoredSortedSet; +import org.redisson.api.RScript; import org.redisson.api.RSet; import org.redisson.api.RedissonClient; +import org.redisson.client.RedisException; +import org.redisson.client.RedisOutOfMemoryException; +import org.redisson.client.codec.StringCodec; +import org.redisson.client.protocol.Decoder; import org.redisson.config.Config; import org.redisson.config.SingleServerConfig; import org.slf4j.Logger; @@ -238,6 +253,511 @@ void testDurationMetric() { testing, instrumentationName.get(), expectedKeys.toArray(new AttributeKey[0])); } + @Test + void wrongTypeErrorIsRecordedInSpanAndMetrics() throws IOException, InterruptedException { + assertThat( + redisServer.execInContainer("redis-cli", "lpush", "wrong-type", "value").getExitCode()) + .isZero(); + assertThatThrownBy(() -> redisson.getBucket("wrong-type").get()) + .isInstanceOf(RedisException.class); + + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(0).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> + span.hasName("GET " + address) + .hasKind(CLIENT) + .hasAttributesSatisfyingExactly( + equalTo(NETWORK_PEER_ADDRESS, ip), + equalTo(NETWORK_PEER_PORT, port), + equalTo(SERVER_ADDRESS, host), + equalTo(SERVER_PORT, port), + equalTo(DB_SYSTEM_NAME, REDIS), + equalTo(DB_NAMESPACE, dbNamespace("0")), + equalTo(DB_QUERY_TEXT, "GET wrong-type"), + equalTo(DB_OPERATION_NAME, "GET"), + equalTo(ERROR_TYPE, "WRONGTYPE"))); + }); + + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram.hasPointsSatisfying( + point -> + point.hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, "WRONGTYPE"))))); + } + + @Test + void serverErrorPrefixContract() { + assertScriptErrorType("OOM command not allowed", "OOM", RedisOutOfMemoryException.class); + assertScriptErrorType("Custom.Code! diagnostic details", "Custom.Code!", null); + assertScriptErrorType("Préfix diagnostic", "Préfix", null); + assertScriptErrorType("", RedisException.class.getName(), null); + + Object result = + redisson.getScript().eval(RScript.Mode.READ_ONLY, "return 1", RScript.ReturnType.VALUE); + assertThat(result).isEqualTo(1L); + assertErrorTypeOnSpanAndMetric(null); + } + + @Test + void reentrantFailureCallbackKeepsClientErrorFallback() + throws InterruptedException, ReflectiveOperationException { + Assumptions.assumeTrue(isClassAvailable("org.redisson.config.CommandMapper")); + + Class commandMapperType = loadCommandMapperType(); + Config config = createConfig(0, null); + config.useSingleServer().setConnectionPoolSize(1).setConnectionMinimumIdleSize(1); + + CountDownLatch evalMappingStarted = new CountDownLatch(1); + CountDownLatch allowEvalMapping = new CountDownLatch(1); + AtomicBoolean interceptCommands = new AtomicBoolean(); + AtomicBoolean failureCallbackActive = new AtomicBoolean(); + AtomicBoolean getMappedInFailureCallback = new AtomicBoolean(); + Object commandMapper = + Proxy.newProxyInstance( + commandMapperType.getClassLoader(), + new Class[] {commandMapperType}, + (proxy, method, args) -> { + if (method.getName().equals("map")) { + String command = (String) args[0]; + if (!interceptCommands.get()) { + return command; + } + if (command.equals("EVAL") || command.equals("EVALSHA")) { + evalMappingStarted.countDown(); + try { + if (!allowEvalMapping.await(30, SECONDS)) { + throw new RedisException("timed out waiting to release command mapping"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RedisException("interrupted while mapping command", e); + } + } else if (command.equals("GET")) { + getMappedInFailureCallback.set(failureCallbackActive.get()); + throw new RedisException("client-side command mapper failure"); + } + return command; + } + return null; + }); + setCommandMapper(config, commandMapperType, commandMapper); + + RedissonClient callbackClient = Redisson.create(config); + interceptCommands.set(true); + testing.clearData(); + try { + testing.runWithSpan( + "parent", + () -> { + RFuture serverError = + callbackClient + .getScript() + .evalAsync( + RScript.Mode.READ_WRITE, + "return redis.error_reply('Custom.Callback! diagnostic details')", + RScript.ReturnType.VALUE); + assertThat(evalMappingStarted.await(30, SECONDS)).isTrue(); + + AtomicBoolean failureCallbackCalled = new AtomicBoolean(); + AtomicReference> clientError = new AtomicReference<>(); + CompletionStage failureCallback = + serverError.whenComplete( + (result, error) -> { + failureCallbackCalled.set(true); + failureCallbackActive.set(true); + try { + clientError.set( + callbackClient.getBucket("callback-client-error").getAsync()); + } finally { + failureCallbackActive.set(false); + } + }); + allowEvalMapping.countDown(); + + assertThatThrownBy(() -> failureCallback.toCompletableFuture().join()) + .hasRootCauseInstanceOf(RedisException.class); + assertThat(failureCallbackCalled).isTrue(); + assertThat(getMappedInFailureCallback).isTrue(); + assertThat((Object) clientError.get()).isNotNull(); + assertThatThrownBy(() -> clientError.get().toCompletableFuture().join()) + .hasRootCauseInstanceOf(RedisException.class); + callbackClient.getBucket("callback-success").set("success"); + }); + } finally { + allowEvalMapping.countDown(); + callbackClient.shutdown(); + } + + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(1).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> span.hasName("parent").hasKind(INTERNAL).hasNoParent(), + span -> + span.hasName("EVAL " + address) + .hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, "Custom.Callback!")), + span -> + span.hasName("GET " + address) + .hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, RedisException.class.getName())), + span -> + span.hasName("SET " + address) + .hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()).doesNotContainKey(ERROR_TYPE))); + }); + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram + .satisfies(data -> assertThat(data.getPoints()).hasSize(3)) + .hasPointsSatisfying( + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, "Custom.Callback!")), + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry( + ERROR_TYPE, + RedisException.class.getName())), + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .doesNotContainKey(ERROR_TYPE))))); + } + + private static Class loadCommandMapperType() throws ClassNotFoundException { + return Class.forName( + "org.redisson.config.CommandMapper", + false, + AbstractRedissonClientTest.class.getClassLoader()); + } + + private static void setCommandMapper( + Config config, Class commandMapperType, Object commandMapper) + throws ReflectiveOperationException { + config + .getClass() + .getMethod("setCommandMapper", commandMapperType) + .invoke(config, commandMapper); + } + + @Test + void failedBatchErrorPrefixIsRecordedInSpanAndMetrics() throws ReflectiveOperationException { + RBatch batch = createBatch(redisson); + batch + .getScript() + .evalAsync( + RScript.Mode.READ_WRITE, + "return redis.error_reply('Batch.Code! diagnostic')", + RScript.ReturnType.VALUE); + + assertThatThrownBy(() -> invokeExecute(batch)).hasRootCauseInstanceOf(RedisException.class); + + assertErrorTypeOnSpanAndMetric("Batch.Code!"); + } + + @Test + void batchRedirectIsRetriedAndPreservesBatchResult() + throws IOException, InterruptedException, ReflectiveOperationException { + Assumptions.assumeTrue(isClassAvailable("org.redisson.api.BatchOptions")); + GenericContainer redirectServer = + new GenericContainer<>("redis:6.2.3-alpine").withExposedPorts(6379); + redirectServer.start(); + try { + String key = "batch-redirect-" + UUID.randomUUID(); + assertThat(redirectServer.execInContainer("redis-cli", "incr", key).getExitCode()).isZero(); + String script = + "local count = redis.call('INCR', '" + + key + + "'); if count == 1 then return redis.error_reply('MOVED 1234 " + + redirectServer.getHost() + + ":" + + redirectServer.getMappedPort(6379) + + "'); end; return count"; + RBatch batch = createBatch(redisson); + batch.getScript().evalAsync(RScript.Mode.READ_WRITE, script, RScript.ReturnType.VALUE); + + assertThat(batchResponses(invokeExecute(batch))).isEqualTo(asList(2L)); + assertRedirectErrorTypeOnAttempts(); + assertFollowingOperationsHaveNoLeakedServerError(); + } finally { + redirectServer.stop(); + } + } + + @Test + void batchRetryErrorPrefixIsCapturedBeforeCompletion() + throws IOException, InterruptedException, ReflectiveOperationException { + Assumptions.assumeTrue(isClassAvailable("org.redisson.api.BatchOptions")); + Assumptions.assumeTrue(isClassAvailable("org.redisson.client.RedisRetryException")); + Class redisRetryException; + try { + redisRetryException = Class.forName("org.redisson.client.RedisRetryException"); + } catch (ClassNotFoundException ignored) { + Assumptions.abort(); + return; + } + assertBatchServerErrorPrefix("LOADING Redis is loading", "LOADING", redisRetryException); + assertFollowingOperationsHaveNoLeakedServerError(); + } + + @Test + void clientSideDecodeFailureKeepsExceptionClassFallback() + throws IOException, InterruptedException { + assertClientSideDecodeFailure("client-side-decode-error"); + } + + private void assertBatchServerErrorPrefix( + String serverError, String expectedErrorType, Class expectedFailureType) + throws ReflectiveOperationException { + RBatch batch = createBatchWithoutRetries(); + batch + .getScript() + .evalAsync( + RScript.Mode.READ_WRITE, + "return redis.error_reply('" + serverError + "')", + RScript.ReturnType.VALUE); + + assertThatThrownBy(() -> invokeExecute(batch)) + .satisfies(error -> assertThat(rootCause(error)).isInstanceOf(expectedFailureType)); + assertErrorTypeOnSpanAndMetric(expectedErrorType); + } + + private RBatch createBatchWithoutRetries() throws ReflectiveOperationException { + Class batchOptions = Class.forName("org.redisson.api.BatchOptions"); + Object options = batchOptions.getMethod("defaults").invoke(null); + batchOptions.getMethod("retryAttempts", int.class).invoke(options, 0); + return (RBatch) + redisson.getClass().getMethod("createBatch", batchOptions).invoke(redisson, options); + } + + private static List batchResponses(Object result) throws ReflectiveOperationException { + if (result instanceof List) { + return (List) result; + } + return (List) result.getClass().getMethod("getResponses").invoke(result); + } + + private void assertFollowingOperationsHaveNoLeakedServerError() + throws IOException, InterruptedException { + redisson.getBucket("after-batch-server-error").set("success"); + assertErrorTypeOnSpanAndMetric(null); + assertClientSideDecodeFailure("client-side-after-batch-error"); + } + + private void assertClientSideDecodeFailure(String key) throws IOException, InterruptedException { + assertThat(redisServer.execInContainer("redis-cli", "set", key, "value").getExitCode()) + .isZero(); + RBucket bucket = + redisson.getBucket( + key, + new StringCodec() { + @Override + public Decoder getValueDecoder() { + return (buffer, state) -> { + throw new RedisException("ERR client-side decode failure"); + }; + } + }); + assertThatThrownBy(bucket::get).isInstanceOf(RedisException.class); + + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(0).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> + span.hasName("GET " + address) + .hasKind(CLIENT) + .hasAttributesSatisfyingExactly( + equalTo(NETWORK_PEER_ADDRESS, ip), + equalTo(NETWORK_PEER_PORT, port), + equalTo(SERVER_ADDRESS, host), + equalTo(SERVER_PORT, port), + equalTo(DB_SYSTEM_NAME, REDIS), + equalTo(DB_NAMESPACE, dbNamespace("0")), + equalTo(DB_QUERY_TEXT, "GET " + key), + equalTo(DB_OPERATION_NAME, "GET"), + equalTo(ERROR_TYPE, RedisException.class.getName()))); + }); + + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram.hasPointsSatisfying( + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry( + ERROR_TYPE, RedisException.class.getName()))))); + testing.clearData(); + } + + private static Throwable rootCause(Throwable error) { + while (error.getCause() != null && error.getCause() != error) { + error = error.getCause(); + } + return error; + } + + private static boolean isClassAvailable(String name) { + try { + Class.forName(name, false, AbstractRedissonClientTest.class.getClassLoader()); + return true; + } catch (ClassNotFoundException ignored) { + return false; + } + } + + private void assertScriptErrorType( + String serverError, String expectedErrorType, Class expectedCause) { + String script = "return redis.error_reply('" + serverError + "')"; + assertThatThrownBy( + () -> + redisson + .getScript() + .eval(RScript.Mode.READ_WRITE, script, RScript.ReturnType.VALUE)) + .satisfies( + error -> { + if (expectedCause != null) { + assertThat(error) + .satisfiesAnyOf( + e -> assertThat(e).isInstanceOf(expectedCause), + e -> assertThat(e).hasRootCauseInstanceOf(expectedCause)); + } + }); + assertErrorTypeOnSpanAndMetric(expectedErrorType); + } + + private static void assertErrorTypeOnSpanAndMetric(String expectedErrorType) { + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(0).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> + span.hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> { + if (expectedErrorType == null) { + assertThat(attributes.asMap()).doesNotContainKey(ERROR_TYPE); + } else { + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, expectedErrorType); + } + })); + }); + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram.hasPointsSatisfying( + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> { + if (expectedErrorType == null) { + assertThat(attributes.asMap()) + .doesNotContainKey(ERROR_TYPE); + } else { + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, expectedErrorType); + } + })))); + testing.clearData(); + } + + private static void assertRedirectErrorTypeOnAttempts() { + AtomicReference instrumentationName = new AtomicReference<>(); + testing.waitAndAssertTraces( + trace -> { + instrumentationName.set(trace.getSpan(0).getInstrumentationScopeInfo().getName()); + trace.hasSpansSatisfyingExactly( + span -> + span.hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()).containsEntry(ERROR_TYPE, "MOVED"))); + }, + trace -> + trace.hasSpansSatisfyingExactly( + span -> + span.hasKind(CLIENT) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()).doesNotContainKey(ERROR_TYPE)))); + testing.waitAndAssertMetrics( + instrumentationName.get(), + metrics -> + metrics + .hasName("db.client.operation.duration") + .hasHistogramSatisfying( + histogram -> + histogram.hasPointsSatisfying( + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .containsEntry(ERROR_TYPE, "MOVED")), + point -> + point + .hasCount(1) + .hasAttributesSatisfying( + attributes -> + assertThat(attributes.asMap()) + .doesNotContainKey(ERROR_TYPE))))); + testing.clearData(); + } + @Test void configuredDatabaseIndex() throws InvocationTargetException, IllegalAccessException { RedissonClient databaseOne = Redisson.create(createConfig(1, null)); @@ -433,8 +953,8 @@ private static Stream batchScenarios() { .build())); } - private static void invokeExecute(RBatch batch) throws ReflectiveOperationException { - batch.getClass().getMethod("execute").invoke(batch); + private static Object invokeExecute(RBatch batch) throws ReflectiveOperationException { + return batch.getClass().getMethod("execute").invoke(batch); } @Test