Skip to content

Share the Lettuce batch span among all commands of the batch #845

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wu-sheng merged 1 commit into from
Nov 1, 2026
Merged

Share the Lettuce batch span among all commands of the batch #845

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGES.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
225 changes: 148 additions & 77 deletions ...n/java/org/apache/skywalking/apm/plugin/lettuce/common/RedisChannelWriterInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<RedisCommand<?, ?, ?>>) 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<RedisCommand<?, ?, ?>> 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");
Expand All @@ -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) {
Expand Down Expand Up @@ -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<RedisCommand<?, ?, ?>>) 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();
}
}
}
Expand All @@ -174,46 +262,29 @@ 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<RedisCommand<?, ?, ?>>) 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());
}

/**
* The CommandHandler checks the written command, while a wrapper of the span carrier, such as an AsyncCommand
* 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;
}
Expand Down
19 changes: 7 additions & 12 deletions .../org/apache/skywalking/apm/plugin/lettuce/common/RedisCommandCancelMethodInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,39 +18,34 @@

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;

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) {
}

@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;
}

@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);
}
}
}
15 changes: 7 additions & 8 deletions ...walking/apm/plugin/lettuce/common/RedisCommandCompleteExceptionallyMethodInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -33,21 +32,21 @@ 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;
}

@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);
}
}
}
Loading
Loading