diff --git a/CHANGES.md b/CHANGES.md index 9a400d92d3..0836436292 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -13,6 +13,10 @@ Release Notes. `Lettuce/BATCH_WRITE` span. Stop the Lettuce plugin from starting a trace of its own for every reactive command created without an active trace, and from creating a span that is never finished for a command that is already cancelled or timed out when it is written. +* Share the `Lettuce/BATCH_WRITE` span among all commands of a batch: it ends when the last of them completes, and is an + error when any of them fails. A batch of a cluster connection gets one span per slot, as Lettuce writes the commands + of each slot as a batch of their own. Mark the span of a reactive Lettuce command as an error when Redis replies with + an error, such as `WRONGTYPE`. All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/267?closed=1) diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java index d34a809995..aa6570513e 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java @@ -25,6 +25,7 @@ import io.lettuce.core.protocol.RedisCommand; import org.apache.skywalking.apm.agent.core.conf.Constants; import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; import org.apache.skywalking.apm.agent.core.context.tag.Tags; import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; @@ -45,63 +46,115 @@ public class RedisChannelWriterInterceptor implements InstanceMethodsAroundInter private static final String PASSWORD_MASK = "******"; private static final String ABBR = "..."; private static final String AUTH = "AUTH"; + private static final String BATCH_WRITE = "BATCH_WRITE"; + /** + * The writer of a cluster connection, which splits a batch by the slots of its commands, and writes the commands of + * each slot to the connection of its node, where they are traced as a batch of their own. + */ + private static final String CLUSTER_DISTRIBUTION_CHANNEL_WRITER = "io.lettuce.core.cluster.ClusterDistributionChannelWriter"; private static final StringCodec STRING_CODEC = new StringCodec(); @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInterceptResult result) { String peer = (String) objInst.getSkyWalkingDynamicField(); - RedisCommand writtenCommand = getWrittenCommand(allArguments[0]); - RedisCommand spanCarrierCommand = getSpanCarrierCommand(writtenCommand); - if (spanCarrierCommand == null) { + if (allArguments[0] instanceof RedisCommand) { + beforeWrite(peer, (RedisCommand) allArguments[0]); + } else if (allArguments[0] instanceof Collection && !isClusterDistributionChannelWriter(objInst)) { + beforeWrite(peer, (Collection>) allArguments[0]); + } + } + + private void beforeWrite(String peer, RedisCommand command) { + RedisCommand spanCarrierCommand = getSpanCarrierCommand(command); + if (!(spanCarrierCommand instanceof EnhancedInstance)) { + return; + } + // command has been handle by another channel writer (cluster or sentinel case) + RedisWriteSpan existingWriteSpan = RedisCommandEnhanceInfo.getWriteSpan((EnhancedInstance) spanCarrierCommand); + if (existingWriteSpan != null) { + //set peer in last channel writer (delegate) + if (peer != null) { + existingWriteSpan.setPeer(peer); + } return; } // A command that is already done, such as one cancelled or timed out while its write was delayed, is dropped // by the CommandHandler, or gets no terminal call that reaches the span carrier, so its span would never be // finished. - if (isDone(writtenCommand, spanCarrierCommand)) { + if (isDone(command, spanCarrierCommand)) { return; } - EnhancedInstance enhancedCommand = (EnhancedInstance) spanCarrierCommand; - RedisCommandEnhanceInfo redisCommandEnhanceInfo = (RedisCommandEnhanceInfo) enhancedCommand.getSkyWalkingDynamicField(); - - if (redisCommandEnhanceInfo == null) { - redisCommandEnhanceInfo = new RedisCommandEnhanceInfo(); + String commandName = getCommandName(command.getType()); + String key = Constants.EMPTY_STRING; + if (LettucePluginConfig.Plugin.Lettuce.TRACE_REDIS_PARAMETERS) { + key = getArgsKey(command); + } + RedisWriteSpan writeSpan = createWriteSpan(peer, commandName, key, getSnapshot(spanCarrierCommand)); + if (!traceCommand(writeSpan, command, spanCarrierCommand)) { + writeSpan.errorOccurred(); } + writeSpan.finishCommand(); + } - // command has been handle by another channel writer (cluster or sentinel case) - AbstractSpan existingSpan = redisCommandEnhanceInfo.getSpan(); - if (existingSpan != null) { - //set peer in last channel writer (delegate) - if (peer != null) { - existingSpan.setPeer(peer); + /** + * The commands of a batch are written to the channel and flushed together, so they share one span, which is + * finished when the last of them is finished. A command that is already done is dropped by the CommandHandler, so + * it has no share of the span. + */ + private void beforeWrite(String peer, Collection> commands) { + boolean traced = false; + RedisCommand lastCommandToTrace = null; + for (RedisCommand command : commands) { + RedisCommand spanCarrierCommand = getSpanCarrierCommand(command); + if (!(spanCarrierCommand instanceof EnhancedInstance)) { + continue; + } + RedisWriteSpan existingWriteSpan = RedisCommandEnhanceInfo.getWriteSpan((EnhancedInstance) spanCarrierCommand); + if (existingWriteSpan != null) { + // The batch has been traced by another channel writer, which delegates the write to this one. + traced = true; + if (peer != null) { + existingWriteSpan.setPeer(peer); + } + } else if (!isDone(command, spanCarrierCommand)) { + lastCommandToTrace = command; } + } + if (traced || lastCommandToTrace == null) { return; } - String operationName = "Lettuce/"; - String key = Constants.EMPTY_STRING; - String command = Constants.EMPTY_STRING; - if (allArguments[0] instanceof RedisCommand) { - RedisCommand redisCommand = (RedisCommand) allArguments[0]; - command = getCommandName(redisCommand.getType()); - operationName = operationName + command; - if (LettucePluginConfig.Plugin.Lettuce.TRACE_REDIS_PARAMETERS) { - key = getArgsKey(redisCommand); + RedisWriteSpan writeSpan = createWriteSpan( + peer, BATCH_WRITE, Constants.EMPTY_STRING, getSnapshot(getSpanCarrierCommand(lastCommandToTrace))); + boolean anyCommandTraced = false; + for (RedisCommand command : commands) { + RedisCommand spanCarrierCommand = getSpanCarrierCommand(command); + if (spanCarrierCommand instanceof EnhancedInstance && !isDone(command, spanCarrierCommand) + && traceCommand(writeSpan, command, spanCarrierCommand)) { + anyCommandTraced = true; } - } else if (allArguments[0] instanceof Collection) { - operationName = operationName + "BATCH_WRITE"; - command = "BATCH_WRITE"; } + // Every command of the batch was cancelled or timed out before it got its share of the span. + if (!anyCommandTraced) { + writeSpan.errorOccurred(); + } + writeSpan.finishCommand(); + } - if (redisCommandEnhanceInfo.getSnapshot() != null) { + /** + * @return the exit span of a write, with the share of the writer, which keeps the span unfinished until every + * command of the write has its share. The writer must finish its share at last. + */ + private RedisWriteSpan createWriteSpan(String peer, String command, String key, ContextSnapshot snapshot) { + if (snapshot != null) { AbstractSpan localSpan = ContextManager.createLocalSpan("RedisReactive/local"); localSpan.setComponent(ComponentsDefine.LETTUCE); SpanLayer.asCache(localSpan); - ContextManager.continued(redisCommandEnhanceInfo.getSnapshot()); + ContextManager.continued(snapshot); } - AbstractSpan span = ContextManager.createExitSpan(operationName, peer); + AbstractSpan span = ContextManager.createExitSpan("Lettuce/" + command, peer); span.setComponent(ComponentsDefine.LETTUCE); Tags.CACHE_TYPE.set(span, "Redis"); @@ -114,22 +167,47 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr span.prepareForAsync(); ContextManager.stopSpan(); - if (redisCommandEnhanceInfo.getSnapshot() != null) { + if (snapshot != null) { ContextManager.stopSpan(); } - - enhancedCommand.setSkyWalkingDynamicField(redisCommandEnhanceInfo.setSpan(span)); + return new RedisWriteSpan(span); + } + + /** + * Gives the command a share of the write span, which its terminal interceptor finishes. + * + * @return false if the command did not keep its share, because it already has a span, or it was cancelled or + * timed out before it got the share, when its terminal interceptor had no span to finish. + */ + private static boolean traceCommand(RedisWriteSpan writeSpan, RedisCommand command, + RedisCommand spanCarrierCommand) { + EnhancedInstance enhancedCommand = (EnhancedInstance) spanCarrierCommand; + RedisCommandEnhanceInfo enhanceInfo = (RedisCommandEnhanceInfo) enhancedCommand.getSkyWalkingDynamicField(); + if (enhanceInfo == null) { + enhanceInfo = new RedisCommandEnhanceInfo(); + } else if (enhanceInfo.getWriteSpan() != null) { + // The same command is in the batch more than once. + return false; + } + writeSpan.addCommand(); + enhancedCommand.setSkyWalkingDynamicField(enhanceInfo.setWriteSpan(writeSpan)); - // The command may have been cancelled or timed out before the span was set, when its terminal interceptor - // had no span to finish. - if (isDone(writtenCommand, spanCarrierCommand)) { - AbstractSpan unfinishedSpan = redisCommandEnhanceInfo.takeSpan(); - if (unfinishedSpan != null) { + if (isDone(command, spanCarrierCommand)) { + RedisWriteSpan unfinishedWriteSpan = enhanceInfo.takeWriteSpan(); + if (unfinishedWriteSpan != null) { enhancedCommand.setSkyWalkingDynamicField(null); - unfinishedSpan.errorOccurred(); - unfinishedSpan.asyncFinish(); + unfinishedWriteSpan.errorOccurred(); + unfinishedWriteSpan.finishCommand(); } + return false; } + return true; + } + + private static ContextSnapshot getSnapshot(RedisCommand spanCarrierCommand) { + RedisCommandEnhanceInfo enhanceInfo = + (RedisCommandEnhanceInfo) ((EnhancedInstance) spanCarrierCommand).getSkyWalkingDynamicField(); + return enhanceInfo == null ? null : enhanceInfo.getSnapshot(); } private String getArgsKey(RedisCommand redisCommand) { @@ -158,14 +236,24 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - RedisCommand redisCommand = getSpanCarrierCommand(allArguments[0]); - if (redisCommand instanceof EnhancedInstance) { - EnhancedInstance enhancedRedisCommand = (EnhancedInstance) redisCommand; - AbstractSpan abstractSpan = RedisCommandEnhanceInfo.takeSpan(enhancedRedisCommand); - if (abstractSpan != null) { + if (allArguments[0] instanceof RedisCommand) { + finishFailedWrite((RedisCommand) allArguments[0], t); + } else if (allArguments[0] instanceof Collection && !isClusterDistributionChannelWriter(objInst)) { + for (RedisCommand command : (Collection>) allArguments[0]) { + finishFailedWrite(command, t); + } + } + } + + private static void finishFailedWrite(RedisCommand command, Throwable t) { + RedisCommand spanCarrierCommand = getSpanCarrierCommand(command); + if (spanCarrierCommand instanceof EnhancedInstance) { + EnhancedInstance enhancedRedisCommand = (EnhancedInstance) spanCarrierCommand; + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.takeWriteSpan(enhancedRedisCommand); + if (writeSpan != null) { enhancedRedisCommand.setSkyWalkingDynamicField(null); - abstractSpan.log(t); - abstractSpan.asyncFinish(); + writeSpan.log(t); + writeSpan.finishCommand(); } } } @@ -174,29 +262,8 @@ protected String getCommandName(ProtocolKeyword protocol) { return "UNKNOWN"; } - /** - * @return the written command. For a batch write, it is the last command that is not done, which carries the span - * of the whole batch, because a done command is dropped while the others are still sent. A command of the batch - * that is cancelled while the batch is being written can still leave it without a span, or with two. - */ - private static RedisCommand getWrittenCommand(Object o) { - if (o instanceof RedisCommand) { - return (RedisCommand) o; - } - if (!(o instanceof Collection)) { - return null; - } - RedisCommand lastCommand = null; - RedisCommand lastCommandNotDone = null; - boolean anyCommandNotDone = false; - for (RedisCommand redisCommand : (Collection>) o) { - lastCommand = redisCommand; - if (redisCommand == null || !isDone(redisCommand, getSpanCarrierCommand(redisCommand))) { - lastCommandNotDone = redisCommand; - anyCommandNotDone = true; - } - } - return anyCommandNotDone ? lastCommandNotDone : lastCommand; + private static boolean isClusterDistributionChannelWriter(EnhancedInstance channelWriter) { + return CLUSTER_DISTRIBUTION_CHANNEL_WRITER.equals(channelWriter.getClass().getName()); } /** @@ -204,16 +271,20 @@ protected String getCommandName(ProtocolKeyword protocol) { * whose cancellation cancels the carrier first, or a ClusterCommand, can look not done after the carrier is. */ private static boolean isDone(RedisCommand writtenCommand, RedisCommand spanCarrierCommand) { - return writtenCommand.isDone() || spanCarrierCommand != null && spanCarrierCommand.isDone(); + return writtenCommand.isDone() || spanCarrierCommand.isDone(); } + /** + * @return the command that carries the span of a written command, which is the command decorated by it, if any. + */ static RedisCommand getSpanCarrierCommand(Object o) { - RedisCommand command = getWrittenCommand(o); - if (command instanceof DecoratedCommand) { - while (command instanceof DecoratedCommand) { - DecoratedCommand wrapper = (DecoratedCommand) command; - command = wrapper.getDelegate(); - } + if (!(o instanceof RedisCommand)) { + return null; + } + RedisCommand command = (RedisCommand) o; + while (command instanceof DecoratedCommand) { + DecoratedCommand wrapper = (DecoratedCommand) command; + command = wrapper.getDelegate(); } return command; } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java index 9bbc7962cd..5c66744bb3 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java @@ -18,8 +18,6 @@ package org.apache.skywalking.apm.plugin.lettuce.common; -import org.apache.skywalking.apm.agent.core.context.tag.StringTag; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; @@ -27,8 +25,6 @@ import java.lang.reflect.Method; public class RedisCommandCancelMethodInterceptor implements InstanceMethodsAroundInterceptor { - private static final String CANCEL_SIGNAL_TAG = "signalType"; - private static final String COMMAND_CANCEL_VALUE = "cancel"; @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, MethodInterceptResult result) { @@ -36,11 +32,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) { - AbstractSpan span = RedisCommandEnhanceInfo.takeSpan(objInst); - if (span != null) { - span.errorOccurred(); - span.tag(new StringTag(CANCEL_SIGNAL_TAG), COMMAND_CANCEL_VALUE); - span.asyncFinish(); + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.takeWriteSpan(objInst); + if (writeSpan != null) { + writeSpan.cancelled(); + writeSpan.finishCommand(); objInst.setSkyWalkingDynamicField(null); } return ret; @@ -48,9 +43,9 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); - if (span != null) { - span.log(t); + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.getWriteSpan(objInst); + if (writeSpan != null) { + writeSpan.log(t); } } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java index 4cf608844b..8f255a3367 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java @@ -18,7 +18,6 @@ package org.apache.skywalking.apm.plugin.lettuce.common; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; @@ -33,11 +32,11 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) { - AbstractSpan span = RedisCommandEnhanceInfo.takeSpan(objInst); - if (span != null) { + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.takeWriteSpan(objInst); + if (writeSpan != null) { Throwable t = (Throwable) allArguments[0]; - span.log(t); - span.asyncFinish(); + writeSpan.log(t); + writeSpan.finishCommand(); objInst.setSkyWalkingDynamicField(null); } return ret; @@ -45,9 +44,9 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); - if (span != null) { - span.log(t); + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.getWriteSpan(objInst); + if (writeSpan != null) { + writeSpan.log(t); } } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java index 76fcc888e9..d2de3eb386 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCompleteMethodInterceptor.java @@ -18,7 +18,8 @@ package org.apache.skywalking.apm.plugin.lettuce.common; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import io.lettuce.core.output.CommandOutput; +import io.lettuce.core.protocol.RedisCommand; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult; @@ -33,9 +34,15 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) { - AbstractSpan span = RedisCommandEnhanceInfo.takeSpan(objInst); - if (span != null) { - span.asyncFinish(); + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.takeWriteSpan(objInst); + if (writeSpan != null) { + // A reactive command is completed without an exception on an error reply, such as WRONGTYPE, while an + // AsyncCommand completes its command exceptionally before. + CommandOutput output = ((RedisCommand) objInst).getOutput(); + if (output != null && output.hasError()) { + writeSpan.errorReply(String.valueOf(output.getError())); + } + writeSpan.finishCommand(); objInst.setSkyWalkingDynamicField(null); } return ret; @@ -44,9 +51,9 @@ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allA @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Throwable t) { - AbstractSpan span = RedisCommandEnhanceInfo.getSpan(objInst); - if (span != null) { - span.log(t); + RedisWriteSpan writeSpan = RedisCommandEnhanceInfo.getWriteSpan(objInst); + if (writeSpan != null) { + writeSpan.log(t); } } } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java index e4b147eb8e..f2040ad536 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandEnhanceInfo.java @@ -20,16 +20,15 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.skywalking.apm.agent.core.context.ContextSnapshot; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; /** * RedisCommandEnhanceInfo holds SkyWalking tracing data for Lettuce commands * executed in different asynchronous models. * - *

The {@link AbstractSpan} is used for non-reactive (blocking) commands - * that are executed asynchronously, where the span needs to be created - * at command submission time and finished when the command completes.

+ *

The {@link RedisWriteSpan} is the exit span of the write that sent the + * command, which is created when the command is written and finished when + * the command, and every other command of the same write, completes.

* *

The {@link ContextSnapshot} is used for reactive commands, where the * tracing context is captured from Reactor {@code Context} and later @@ -37,39 +36,39 @@ * boundaries.

*/ class RedisCommandEnhanceInfo { - - private final AtomicReference span = new AtomicReference<>(); + + private final AtomicReference writeSpan = new AtomicReference<>(); private ContextSnapshot snapshot; - public AbstractSpan getSpan() { - return span.get(); + public RedisWriteSpan getWriteSpan() { + return writeSpan.get(); } /** - * @return the span of the command, or null if the command has no span. A reactive command only carries the + * @return the write span of the command, or null if the command has no span. A reactive command only carries the * snapshot until it is written, so a command cancelled before that has no span to finish. */ - static AbstractSpan getSpan(EnhancedInstance command) { + static RedisWriteSpan getWriteSpan(EnhancedInstance command) { RedisCommandEnhanceInfo enhanceInfo = (RedisCommandEnhanceInfo) command.getSkyWalkingDynamicField(); - return enhanceInfo == null ? null : enhanceInfo.getSpan(); + return enhanceInfo == null ? null : enhanceInfo.getWriteSpan(); } /** - * @return the span of the command, which the caller must finish, or null if the command has no span or another - * caller has taken it. A command can be cancelled while its write is traced, so more than one caller may try to - * finish the span, and only the first one gets it. + * @return the write span of the command, whose share of this command the caller must finish, or null if the + * command has no span or another caller has taken it. A command can be cancelled while its write is traced, so more + * than one caller may try to finish its share, and only the first one gets it. */ - static AbstractSpan takeSpan(EnhancedInstance command) { + static RedisWriteSpan takeWriteSpan(EnhancedInstance command) { RedisCommandEnhanceInfo enhanceInfo = (RedisCommandEnhanceInfo) command.getSkyWalkingDynamicField(); - return enhanceInfo == null ? null : enhanceInfo.takeSpan(); + return enhanceInfo == null ? null : enhanceInfo.takeWriteSpan(); } - AbstractSpan takeSpan() { - return span.getAndSet(null); + RedisWriteSpan takeWriteSpan() { + return writeSpan.getAndSet(null); } - public RedisCommandEnhanceInfo setSpan(AbstractSpan span) { - this.span.set(span); + public RedisCommandEnhanceInfo setWriteSpan(RedisWriteSpan writeSpan) { + this.writeSpan.set(writeSpan); return this; } diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisWriteSpan.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisWriteSpan.java new file mode 100644 index 0000000000..38050dfd6a --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/main/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisWriteSpan.java @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.plugin.lettuce.common; + +import java.util.LinkedHashMap; +import java.util.Map; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +import org.apache.skywalking.apm.agent.core.context.status.StatusCheckService; +import org.apache.skywalking.apm.agent.core.context.tag.StringTag; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; + +/** + * The exit span of one write to Redis, shared by all commands written in it. A command written alone has a span of its + * own, while the commands written as a batch share one span. The span is finished when the last of its commands is + * finished, and is an error when any of them fails. Only the first failure is logged, so a batch that fails as a + * whole, such as by a closed connection or by timeouts, does not log one error per command. + * + *

Each command finishes its share once, after taking it by {@link RedisCommandEnhanceInfo#takeWriteSpan()}, so a + * command finished by more than one caller, such as cancelled while it completes, is counted once. The commands of a + * batch can be finished in different threads, such as by the event loop and by a thread that cancels one of them, so + * the span is only changed while holding the lock of this object.

+ */ +class RedisWriteSpan { + + private static final StringTag CANCEL_SIGNAL_TAG = new StringTag("signalType"); + private static final String COMMAND_CANCEL_VALUE = "cancel"; + + private final AbstractSpan span; + /** + * The shares of the commands, and of the writer that creates the span, which keeps the span unfinished until every + * command of the write has its share. + */ + private int unfinishedCommands = 1; + private boolean failureLogged; + + RedisWriteSpan(AbstractSpan span) { + this.span = span; + } + + synchronized void addCommand() { + unfinishedCommands++; + } + + synchronized void setPeer(String peer) { + if (!isFinished()) { + span.setPeer(peer); + } + } + + /** + * A command failed with an exception, such as a closed connection or a timeout. An exception that is not the first + * failure is not logged, but still makes the span an error, unless it is one of the ignored exceptions. + */ + synchronized void log(Throwable t) { + if (isFirstFailure()) { + span.log(t); + } else if (!isFinished() && ServiceManager.INSTANCE.findService(StatusCheckService.class).isError(t)) { + span.errorOccurred(); + } + } + + /** + * A command got an error reply from Redis, such as WRONGTYPE, and was completed without an exception, as a + * reactive command is. + */ + synchronized void errorReply(String error) { + if (isFinished()) { + return; + } + span.errorOccurred(); + if (isFirstFailure()) { + Map event = new LinkedHashMap<>(); + event.put("event", "error"); + event.put("message", error); + span.log(System.currentTimeMillis(), event); + } + } + + synchronized void cancelled() { + if (isFinished()) { + return; + } + span.errorOccurred(); + if (isFirstFailure()) { + span.tag(CANCEL_SIGNAL_TAG, COMMAND_CANCEL_VALUE); + } + } + + synchronized void errorOccurred() { + if (!isFinished()) { + span.errorOccurred(); + } + } + + /** + * Finishes the share of one command, and the span once no command of it is unfinished. + */ + synchronized void finishCommand() { + if (!isFinished() && --unfinishedCommands == 0) { + span.asyncFinish(); + } + } + + private boolean isFinished() { + return unfinishedCommands == 0; + } + + private boolean isFirstFailure() { + if (failureLogged || isFinished()) { + return false; + } + failureLogged = true; + return true; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptorTest.java index 91ae4b4218..c987005392 100644 --- a/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/lettuce-plugins/lettuce-common/src/test/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptorTest.java @@ -23,10 +23,14 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; -import java.lang.reflect.Method; -import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; +import org.apache.skywalking.apm.agent.core.boot.ServiceManager; +import org.apache.skywalking.apm.agent.core.conf.Config; +import org.apache.skywalking.apm.agent.core.context.status.StatusCheckService; import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan; +import org.apache.skywalking.apm.agent.core.context.trace.LogDataEntity; import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment; import org.apache.skywalking.apm.agent.core.context.util.TagValuePair; @@ -46,7 +50,10 @@ import org.mockito.junit.MockitoJUnit; import org.mockito.junit.MockitoRule; import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.codec.StringCodec; import io.lettuce.core.output.CommandOutput; +import io.lettuce.core.output.StatusOutput; +import io.lettuce.core.protocol.AsyncCommand; import io.lettuce.core.protocol.Command; import io.lettuce.core.protocol.CommandArgs; import io.lettuce.core.protocol.CommandType; @@ -111,6 +118,9 @@ public void setSkyWalkingDynamicField(Object value) { } } + private static class IgnoredException extends RuntimeException { + } + private static class CancelledWhileSpanIsCreatedCommand extends MockRedisCommand { public CancelledWhileSpanIsCreatedCommand(ProtocolKeyword type, CommandOutput output) { super(type, output); @@ -195,36 +205,126 @@ public void testSpanIsTakenByOneTerminalInterceptor() { } @Test - public void testGetSpanCarrierCommand() throws Exception { - Command command = new Command<>(CommandType.SET, null, null); - Method getSpanCarrierCommandMethod = interceptor.getClass() - .getDeclaredMethod("getSpanCarrierCommand", Object.class); - getSpanCarrierCommandMethod.setAccessible(true); - - RedisCommand redisCommand = - (RedisCommand) getSpanCarrierCommandMethod.invoke(interceptor, command); - assertEquals(command, redisCommand); - List> list = new ArrayList<>(); - for (int i = 0; i < 10; i++) { - list.add(new Command<>(CommandType.SET, null, null)); + public void testBatchSpanIsFinishedByItsLastUnfinishedCommand() { + MockRedisCommand first = new MockRedisCommand<>(CommandType.SET, null); + MockRedisCommand last = new MockRedisCommand<>(CommandType.SET, null); + interceptor.beforeMethod(mockRedisChannelWriterInstance, null, new Object[]{Arrays.asList(first, last)}, null, null); + + // The last command is cancelled while the first one still waits for its reply. + last.cancel(); + new RedisCommandCancelMethodInterceptor().afterMethod(last, null, null, null, null); + assertThat(segmentStorage.getTraceSegments().size(), is(0)); + + redisCommandCompleteMethodInterceptor.afterMethod(first, null, null, null, null); + AbstractTracingSpan span = getBatchSpan(); + assertThat(SpanHelper.getErrorOccurred(span), is(true)); + assertThat(SpanHelper.getPeer(span), is(PEER)); + } + + @Test + public void testErrorReplyOfEarlierCommandFailsBatchSpan() { + MockRedisCommand errorReply = new MockRedisCommand<>(CommandType.INCR, new StatusOutput<>(StringCodec.UTF8)); + MockRedisCommand last = new MockRedisCommand<>(CommandType.SET, null); + interceptor.beforeMethod(mockRedisChannelWriterInstance, null, new Object[]{Arrays.asList(errorReply, last)}, null, null); + + // A reactive command is completed without an exception on an error reply. + errorReply.getOutput().setError("ERR value is not an integer or out of range"); + redisCommandCompleteMethodInterceptor.afterMethod(errorReply, null, null, null, null); + assertThat(segmentStorage.getTraceSegments().size(), is(0)); + redisCommandCompleteMethodInterceptor.afterMethod(last, null, null, null, null); + + AbstractTracingSpan span = getBatchSpan(); + assertThat(SpanHelper.getErrorOccurred(span), is(true)); + List logs = SpanHelper.getLogs(span); + assertThat(logs.size(), is(1)); + assertThat(logs.get(0).getLogs().get(0).getValue(), is("error")); + assertThat(logs.get(0).getLogs().get(1).getValue(), is("ERR value is not an integer or out of range")); + } + + @Test + public void testFailuresOfBatchCommandsAreLoggedOnce() { + MockRedisCommand first = new MockRedisCommand<>(CommandType.SET, null); + MockRedisCommand last = new MockRedisCommand<>(CommandType.SET, null); + interceptor.beforeMethod(mockRedisChannelWriterInstance, null, new Object[]{Arrays.asList(first, last)}, null, null); + + // The connection is closed while both commands wait for their replies. + new RedisCommandCompleteExceptionallyMethodInterceptor().afterMethod( + first, null, new Object[]{new IllegalStateException("closed")}, null, null); + new RedisCommandCompleteExceptionallyMethodInterceptor().afterMethod( + last, null, new Object[]{new IllegalStateException("closed")}, null, null); + + AbstractTracingSpan span = getBatchSpan(); + assertThat(SpanHelper.getErrorOccurred(span), is(true)); + assertThat(SpanHelper.getLogs(span).size(), is(1)); + } + + @Test + public void testIgnoredFirstExceptionDoesNotHideLaterFailure() throws Throwable { + StatusCheckService statusCheckService = ServiceManager.INSTANCE.findService(StatusCheckService.class); + Config.StatusCheck.IGNORED_EXCEPTIONS = IgnoredException.class.getName(); + statusCheckService.prepare(); + try { + MockRedisCommand first = new MockRedisCommand<>(CommandType.SET, null); + MockRedisCommand last = new MockRedisCommand<>(CommandType.SET, null); + interceptor.beforeMethod(mockRedisChannelWriterInstance, null, new Object[]{Arrays.asList(first, last)}, null, null); + + new RedisCommandCompleteExceptionallyMethodInterceptor().afterMethod( + first, null, new Object[]{new IgnoredException()}, null, null); + new RedisCommandCompleteExceptionallyMethodInterceptor().afterMethod( + last, null, new Object[]{new IllegalStateException("failed")}, null, null); + + AbstractTracingSpan span = getBatchSpan(); + assertThat(SpanHelper.getErrorOccurred(span), is(true)); + assertThat(SpanHelper.getLogs(span).size(), is(1)); + } finally { + Config.StatusCheck.IGNORED_EXCEPTIONS = ""; + statusCheckService.prepare(); } - list.add(command); - RedisCommand last = - (RedisCommand) getSpanCarrierCommandMethod.invoke(interceptor, list); - assertEquals(command, last); - Command cancelled = new Command<>(CommandType.SET, null, null); - cancelled.cancel(); - list.add(cancelled); - RedisCommand lastNotDone = - (RedisCommand) getSpanCarrierCommandMethod.invoke(interceptor, list); - assertEquals(command, lastNotDone); - list.remove(cancelled); - RedisCommand nullValue1 = - (RedisCommand) getSpanCarrierCommandMethod.invoke(interceptor, (Object) null); - assertNull(nullValue1); - list.add(null); - RedisCommand nullValue2 = - (RedisCommand) getSpanCarrierCommandMethod.invoke(interceptor, list); - assertNull(nullValue2); + } + + @Test + public void testCommandCancelledWhileBatchIsTracedLeavesOneSpan() { + MockRedisCommand first = new MockRedisCommand<>(CommandType.SET, null); + CancelledWhileSpanIsCreatedCommand last = new CancelledWhileSpanIsCreatedCommand<>(CommandType.SET, null); + interceptor.beforeMethod(mockRedisChannelWriterInstance, null, new Object[]{Arrays.asList(first, last)}, null, null); + + assertNull(last.getSkyWalkingDynamicField()); + assertThat(segmentStorage.getTraceSegments().size(), is(0)); + redisCommandCompleteMethodInterceptor.afterMethod(first, null, null, null, null); + assertThat(SpanHelper.getErrorOccurred(getBatchSpan()), is(true)); + } + + @Test + public void testDelegateWriterKeepsSpanOfTracedBatch() { + MockInstance outerWriter = new MockInstance(); + MockRedisCommand first = new MockRedisCommand<>(CommandType.SET, null); + MockRedisCommand last = new MockRedisCommand<>(CommandType.SET, null); + List> batch = Arrays.asList(first, last); + interceptor.beforeMethod(outerWriter, null, new Object[]{batch}, null, null); + + // The last command is cancelled before the outer writer delegates the write. + last.cancel(); + new RedisCommandCancelMethodInterceptor().afterMethod(last, null, null, null, null); + interceptor.beforeMethod(mockRedisChannelWriterInstance, null, new Object[]{batch}, null, null); + + redisCommandCompleteMethodInterceptor.afterMethod(first, null, null, null, null); + assertThat(SpanHelper.getPeer(getBatchSpan()), is(PEER)); + } + + @Test + public void testGetSpanCarrierCommand() { + Command command = new Command<>(CommandType.SET, null, null); + assertEquals(command, RedisChannelWriterInterceptor.getSpanCarrierCommand(command)); + assertEquals(command, RedisChannelWriterInterceptor.getSpanCarrierCommand(new AsyncCommand<>(command))); + assertNull(RedisChannelWriterInterceptor.getSpanCarrierCommand(null)); + assertNull(RedisChannelWriterInterceptor.getSpanCarrierCommand(Collections.singletonList(command))); + } + + private AbstractTracingSpan getBatchSpan() { + assertThat(segmentStorage.getTraceSegments().size(), is(1)); + List spans = SegmentHelper.getSpans(segmentStorage.getTraceSegments().get(0)); + assertThat(spans.size(), is(1)); + assertThat(spans.get(0).getOperationName(), is("Lettuce/BATCH_WRITE")); + return spans.get(0); } } diff --git a/test/plugin/scenarios/lettuce-6.5.x-scenario/config/expectedData.yaml b/test/plugin/scenarios/lettuce-6.5.x-scenario/config/expectedData.yaml index 9ba6c04dda..49b5b7c914 100644 --- a/test/plugin/scenarios/lettuce-6.5.x-scenario/config/expectedData.yaml +++ b/test/plugin/scenarios/lettuce-6.5.x-scenario/config/expectedData.yaml @@ -203,6 +203,61 @@ segmentItems: tags: - { key: cache.type, value: Redis } - { key: cache.cmd, value: BATCH_WRITE } + - operationName: Lettuce/BATCH_WRITE + parentSpanId: 0 + spanId: 10 + spanLayer: Cache + startTime: not null + endTime: not null + componentId: 57 + isError: true + spanType: Exit + peer: not null + skipAnalysis: false + tags: + - { key: cache.type, value: Redis } + - { key: cache.cmd, value: BATCH_WRITE } + logs: + - logEvent: + - { key: event, value: error } + - { key: error.kind, value: io.lettuce.core.RedisCommandExecutionException } + - { key: message, value: not null } + - { key: stack, value: not null } + - operationName: Lettuce/Reactive/createMono + parentSpanId: 0 + spanId: 11 + spanLayer: Cache + startTime: not null + endTime: not null + componentId: 57 + isError: false + spanType: Local + skipAnalysis: false + - operationName: Lettuce/INCR + parentSpanId: 12 + spanId: 13 + spanLayer: Cache + startTime: not null + endTime: not null + componentId: 57 + isError: true + spanType: Exit + peer: not null + skipAnalysis: false + tags: + - { key: cache.type, value: Redis } + - { key: cache.key, value: key2 } + - { key: cache.cmd, value: INCR } + - { key: cache.op, value: write } + logs: + - logEvent: + - { key: event, value: error } + - { key: message, value: not null } + - operationName: RedisReactive/local + parentSpanId: 0 + spanId: 12 + isError: false + spanType: Local - operationName: GET:/lettuce-6.5.x-scenario/case/lettuce-case parentSpanId: -1 spanId: 0 diff --git a/test/plugin/scenarios/lettuce-6.5.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java b/test/plugin/scenarios/lettuce-6.5.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java index 45984070e0..359db982b4 100644 --- a/test/plugin/scenarios/lettuce-6.5.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java +++ b/test/plugin/scenarios/lettuce-6.5.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java @@ -26,6 +26,7 @@ import io.lettuce.core.api.reactive.RedisReactiveCommands; import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.codec.StringCodec; +import io.lettuce.core.output.IntegerOutput; import io.lettuce.core.output.StatusOutput; import io.lettuce.core.protocol.AsyncCommand; import io.lettuce.core.protocol.Command; @@ -86,6 +87,18 @@ public String lettuceCase() { connection0.dispatch(Arrays.asList(batchSet0, batchSet1)); LettuceFutures.awaitAll(5, TimeUnit.SECONDS, batchSet0, batchSet1); + // The error reply of the first command makes the span of the batch an error, while the span ends with the + // reply of the last command. Replies come in order, so the first command is done when the last one is. + AsyncCommand batchIncr = incrCommand("key2"); + AsyncCommand batchSet2 = setCommand("key4", "value4"); + connection0.dispatch(Arrays.asList(batchIncr, batchSet2)); + if (!LettuceFutures.awaitAll(5, TimeUnit.SECONDS, batchSet2)) { + throw new IllegalStateException("The batch is not completed"); + } + + // A reactive command is completed without an exception on an error reply, which still makes its span an error. + reactiveCommands.incr("key2").onErrorResume(e -> Mono.empty()).block(); + connection0.close(); connection1.close(); connection2.close(); @@ -98,6 +111,11 @@ private static AsyncCommand setCommand(String key, Strin return new AsyncCommand<>(new Command<>(CommandType.SET, new StatusOutput<>(StringCodec.UTF8), args)); } + private static AsyncCommand incrCommand(String key) { + CommandArgs args = new CommandArgs<>(StringCodec.UTF8).addKey(key); + return new AsyncCommand<>(new Command<>(CommandType.INCR, new IntegerOutput<>(StringCodec.UTF8), args)); + } + @RequestMapping("/healthCheck") @ResponseBody public String healthCheck() { diff --git a/test/plugin/scenarios/lettuce-scenario/config/expectedData.yaml b/test/plugin/scenarios/lettuce-scenario/config/expectedData.yaml index d2831cb893..ca9f7718d6 100644 --- a/test/plugin/scenarios/lettuce-scenario/config/expectedData.yaml +++ b/test/plugin/scenarios/lettuce-scenario/config/expectedData.yaml @@ -203,6 +203,61 @@ segmentItems: tags: - { key: cache.type, value: Redis } - { key: cache.cmd, value: BATCH_WRITE } + - operationName: Lettuce/BATCH_WRITE + parentSpanId: 0 + spanId: 10 + spanLayer: Cache + startTime: not null + endTime: not null + componentId: 57 + isError: true + spanType: Exit + peer: not null + skipAnalysis: false + tags: + - { key: cache.type, value: Redis } + - { key: cache.cmd, value: BATCH_WRITE } + logs: + - logEvent: + - { key: event, value: error } + - { key: error.kind, value: io.lettuce.core.RedisCommandExecutionException } + - { key: message, value: not null } + - { key: stack, value: not null } + - operationName: Lettuce/Reactive/createMono + parentSpanId: 0 + spanId: 11 + spanLayer: Cache + startTime: not null + endTime: not null + componentId: 57 + isError: false + spanType: Local + skipAnalysis: false + - operationName: Lettuce/INCR + parentSpanId: 12 + spanId: 13 + spanLayer: Cache + startTime: not null + endTime: not null + componentId: 57 + isError: true + spanType: Exit + peer: not null + skipAnalysis: false + tags: + - { key: cache.type, value: Redis } + - { key: cache.key, value: key2 } + - { key: cache.cmd, value: INCR } + - { key: cache.op, value: write } + logs: + - logEvent: + - { key: event, value: error } + - { key: message, value: not null } + - operationName: RedisReactive/local + parentSpanId: 0 + spanId: 12 + isError: false + spanType: Local - operationName: GET:/lettuce-scenario/case/lettuce-case parentSpanId: -1 spanId: 0 diff --git a/test/plugin/scenarios/lettuce-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java b/test/plugin/scenarios/lettuce-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java index 45984070e0..359db982b4 100644 --- a/test/plugin/scenarios/lettuce-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java +++ b/test/plugin/scenarios/lettuce-scenario/src/main/java/org/apache/skywalking/apm/testcase/lettuce/controller/LettuceController.java @@ -26,6 +26,7 @@ import io.lettuce.core.api.reactive.RedisReactiveCommands; import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.codec.StringCodec; +import io.lettuce.core.output.IntegerOutput; import io.lettuce.core.output.StatusOutput; import io.lettuce.core.protocol.AsyncCommand; import io.lettuce.core.protocol.Command; @@ -86,6 +87,18 @@ public String lettuceCase() { connection0.dispatch(Arrays.asList(batchSet0, batchSet1)); LettuceFutures.awaitAll(5, TimeUnit.SECONDS, batchSet0, batchSet1); + // The error reply of the first command makes the span of the batch an error, while the span ends with the + // reply of the last command. Replies come in order, so the first command is done when the last one is. + AsyncCommand batchIncr = incrCommand("key2"); + AsyncCommand batchSet2 = setCommand("key4", "value4"); + connection0.dispatch(Arrays.asList(batchIncr, batchSet2)); + if (!LettuceFutures.awaitAll(5, TimeUnit.SECONDS, batchSet2)) { + throw new IllegalStateException("The batch is not completed"); + } + + // A reactive command is completed without an exception on an error reply, which still makes its span an error. + reactiveCommands.incr("key2").onErrorResume(e -> Mono.empty()).block(); + connection0.close(); connection1.close(); connection2.close(); @@ -98,6 +111,11 @@ private static AsyncCommand setCommand(String key, Strin return new AsyncCommand<>(new Command<>(CommandType.SET, new StatusOutput<>(StringCodec.UTF8), args)); } + private static AsyncCommand incrCommand(String key) { + CommandArgs args = new CommandArgs<>(StringCodec.UTF8).addKey(key); + return new AsyncCommand<>(new Command<>(CommandType.INCR, new IntegerOutput<>(StringCodec.UTF8), args)); + } + @RequestMapping("/healthCheck") @ResponseBody public String healthCheck() {