jiulinxiri/rocketmq-dashboard
1
0
Fork 0
mirror of https://github.com/apache/rocketmq-dashboard.git synced 2025-09-11 20:15:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #768] Console message trace compatibility Optimization

Browse Source
This commit is contained in:
StyleTang
2021-08-03 20:13:16 +08:00
committed by GitHub
parent 399b61aee1
commit c9d7cb20e7
4 changed files with 109 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
src/main/java/org/apache/rocketmq/console/service/impl/MessageTraceServiceImpl.java
View File

@@ -55,6 +55,9 @@ public class MessageTraceServiceImpl implements MessageTraceService {
private Logger logger = LoggerFactory.getLogger(MessageTraceServiceImpl.class); private Logger logger = LoggerFactory.getLogger(MessageTraceServiceImpl.class);
private final static int QUERY_MESSAGE_MAX_NUM = 64; private final static int QUERY_MESSAGE_MAX_NUM = 64;
private final static String UNKNOWN_GROUP_NAME = "%UNKNOWN_GROUP%";
private final static int MESSAGE_TRACE_MISSING_VALUE = -1;
@Resource @Resource
private MQAdminExt mqAdminExt; private MQAdminExt mqAdminExt;
@@ -127,7 +130,7 @@ public class MessageTraceServiceImpl implements MessageTraceService {
private TraceNode buildTransactionNode(MessageTraceView messageTraceView) { private TraceNode buildTransactionNode(MessageTraceView messageTraceView) {
TraceNode transactionNode = buildTraceNode(messageTraceView); TraceNode transactionNode = buildTraceNode(messageTraceView);
transactionNode.setCostTime(-1); transactionNode.setCostTime(MESSAGE_TRACE_MISSING_VALUE);
return transactionNode; return transactionNode;
} }
@@ -135,21 +138,9 @@ public class MessageTraceServiceImpl implements MessageTraceService {
Map<String, Pair<MessageTraceView, MessageTraceView>> requestIdTracePairMap) { Map<String, Pair<MessageTraceView, MessageTraceView>> requestIdTracePairMap) {
Map<String, List<TraceNode>> subscriptionTraceNodeMap = Maps.newHashMap(); Map<String, List<TraceNode>> subscriptionTraceNodeMap = Maps.newHashMap();
for (Pair<MessageTraceView, MessageTraceView> traceNodePair : requestIdTracePairMap.values()) { for (Pair<MessageTraceView, MessageTraceView> traceNodePair : requestIdTracePairMap.values()) {
traceNodePair = makeBeforeOrAfterMissCompatible(traceNodePair); List<TraceNode> traceNodeList = subscriptionTraceNodeMap
MessageTraceView subBeforeTrace = traceNodePair.getObject1(); .computeIfAbsent(buildGroupName(traceNodePair), (o) -> Lists.newArrayList());
MessageTraceView subAfterTrace = traceNodePair.getObject2(); traceNodeList.add(buildConsumeMessageTraceNode(traceNodePair));
List<TraceNode> traceNodeList = subscriptionTraceNodeMap.computeIfAbsent(subBeforeTrace.getGroupName(),
(o) -> Lists.newArrayList());
TraceNode consumeNode = new TraceNode();
consumeNode.setRequestId(subBeforeTrace.getRequestId());
consumeNode.setStoreHost(subBeforeTrace.getStoreHost());
consumeNode.setClientHost(subBeforeTrace.getClientHost());
consumeNode.setRetryTimes(subBeforeTrace.getRetryTimes());
consumeNode.setBeginTimestamp(subBeforeTrace.getTimeStamp());
consumeNode.setCostTime(subAfterTrace.getCostTime());
consumeNode.setEndTimestamp(subBeforeTrace.getTimeStamp() + Math.max(0, subAfterTrace.getCostTime()));
consumeNode.setStatus(subAfterTrace.getStatus());
traceNodeList.add(consumeNode);
} }
return subscriptionTraceNodeMap.entrySet().stream() return subscriptionTraceNodeMap.entrySet().stream()
.map((Function<Map.Entry<String, List<TraceNode>>, SubscriptionNode>) subscriptionEntry -> { .map((Function<Map.Entry<String, List<TraceNode>>, SubscriptionNode>) subscriptionEntry -> {
@@ -161,23 +152,57 @@ public class MessageTraceServiceImpl implements MessageTraceService {
}).collect(Collectors.toList()); }).collect(Collectors.toList());
} }
private Pair<MessageTraceView, MessageTraceView> makeBeforeOrAfterMissCompatible(Pair<MessageTraceView, MessageTraceView> traceNodePair) { private <E> E getTraceValue(Pair<MessageTraceView, MessageTraceView> traceNodePair, Function<MessageTraceView, E> function) {
if (traceNodePair.getObject1() != null && traceNodePair.getObject2() != null) { if (traceNodePair.getObject1() != null) {
return traceNodePair; return function.apply(traceNodePair.getObject1());
} }
MessageTraceView subBeforeTrace = traceNodePair.getObject1(); return function.apply(traceNodePair.getObject2());
MessageTraceView subAfterTrace = traceNodePair.getObject2(); }
if (subBeforeTrace == null) {
subBeforeTrace = new MessageTraceView(); private String buildGroupName(Pair<MessageTraceView, MessageTraceView> traceNodePair) {
BeanUtils.copyProperties(subAfterTrace, subBeforeTrace); String groupName = getTraceValue(traceNodePair, MessageTraceView::getGroupName);
if (StringUtils.isNoneBlank(groupName)) {
return groupName;
} }
if (subAfterTrace == null) { return UNKNOWN_GROUP_NAME;
subAfterTrace = new MessageTraceView(); }
BeanUtils.copyProperties(subBeforeTrace, subAfterTrace);
subAfterTrace.setStatus(MessageTraceStatusEnum.UNKNOWN.getStatus()); private TraceNode buildConsumeMessageTraceNode(Pair<MessageTraceView, MessageTraceView> pair) {
subAfterTrace.setCostTime(-1); MessageTraceView subBeforeTrace = pair.getObject1();
MessageTraceView subAfterTrace = pair.getObject2();
TraceNode consumeNode = new TraceNode();
consumeNode.setRequestId(getTraceValue(pair, MessageTraceView::getRequestId));
consumeNode.setStoreHost(getTraceValue(pair, MessageTraceView::getStoreHost));
consumeNode.setClientHost(getTraceValue(pair, MessageTraceView::getClientHost));
if (subBeforeTrace != null) {
consumeNode.setRetryTimes(subBeforeTrace.getRetryTimes());
consumeNode.setBeginTimestamp(subBeforeTrace.getTimeStamp());
} else {
consumeNode.setRetryTimes(MESSAGE_TRACE_MISSING_VALUE);
consumeNode.setBeginTimestamp(MESSAGE_TRACE_MISSING_VALUE);
} }
return new Pair<>(subBeforeTrace, subAfterTrace); if (subAfterTrace != null) {
consumeNode.setCostTime(subAfterTrace.getCostTime());
consumeNode.setStatus(subAfterTrace.getStatus());
if (subAfterTrace.getTimeStamp() > 0) {
consumeNode.setEndTimestamp(subAfterTrace.getTimeStamp());
} else {
if (subBeforeTrace != null) {
if (subAfterTrace.getCostTime() >= 0) {
consumeNode.setEndTimestamp(subBeforeTrace.getTimeStamp() + subAfterTrace.getCostTime());
} else {
consumeNode.setEndTimestamp(subBeforeTrace.getTimeStamp());
}
} else {
consumeNode.setEndTimestamp(MESSAGE_TRACE_MISSING_VALUE);
}
}
} else {
consumeNode.setCostTime(MESSAGE_TRACE_MISSING_VALUE);
consumeNode.setEndTimestamp(MESSAGE_TRACE_MISSING_VALUE);
consumeNode.setStatus(MessageTraceStatusEnum.UNKNOWN.getStatus());
}
return consumeNode;
} }
private void putIntoMessageTraceViewGroupMap(MessageTraceView messageTraceView, private void putIntoMessageTraceViewGroupMap(MessageTraceView messageTraceView,

18
src/main/java/org/apache/rocketmq/console/util/MsgTraceDecodeUtil.java
View File

@@ -52,7 +52,7 @@ public class MsgTraceDecodeUtil {
for (String context : contextList) { for (String context : contextList) {
String[] line = context.split(String.valueOf(TraceConstants.CONTENT_SPLITOR)); String[] line = context.split(String.valueOf(TraceConstants.CONTENT_SPLITOR));
if (line[0].equals(Pub.name())) { if (line[0].equals(Pub.name())) {
TraceContext pubContext = new TraceContext(); TraceContext pubContext = initTraceContext();
pubContext.setTraceType(Pub); pubContext.setTraceType(Pub);
pubContext.setTimeStamp(Long.parseLong(line[1])); pubContext.setTimeStamp(Long.parseLong(line[1]));
pubContext.setRegionId(line[2]); pubContext.setRegionId(line[2]);
@@ -94,7 +94,7 @@ public class MsgTraceDecodeUtil {
pubContext.getTraceBeans().add(bean); pubContext.getTraceBeans().add(bean);
resList.add(pubContext); resList.add(pubContext);
} else if (line[0].equals(TraceType.SubBefore.name())) { } else if (line[0].equals(TraceType.SubBefore.name())) {
TraceContext subBeforeContext = new TraceContext(); TraceContext subBeforeContext = initTraceContext();
subBeforeContext.setTraceType(TraceType.SubBefore); subBeforeContext.setTraceType(TraceType.SubBefore);
subBeforeContext.setTimeStamp(Long.parseLong(line[1])); subBeforeContext.setTimeStamp(Long.parseLong(line[1]));
subBeforeContext.setRegionId(line[2]); subBeforeContext.setRegionId(line[2]);
@@ -108,7 +108,7 @@ public class MsgTraceDecodeUtil {
subBeforeContext.getTraceBeans().add(bean); subBeforeContext.getTraceBeans().add(bean);
resList.add(subBeforeContext); resList.add(subBeforeContext);
} else if (line[0].equals(TraceType.SubAfter.name())) { } else if (line[0].equals(TraceType.SubAfter.name())) {
TraceContext subAfterContext = new TraceContext(); TraceContext subAfterContext = initTraceContext();
subAfterContext.setTraceType(TraceType.SubAfter); subAfterContext.setTraceType(TraceType.SubAfter);
subAfterContext.setRequestId(line[1]); subAfterContext.setRequestId(line[1]);
TraceBean bean = new TraceBean(); TraceBean bean = new TraceBean();
@@ -128,16 +128,18 @@ public class MsgTraceDecodeUtil {
case TRACE_MSG_SUBAFTER_V3_LEN: case TRACE_MSG_SUBAFTER_V3_LEN:
subAfterContext.setContextCode(Integer.parseInt(line[6])); subAfterContext.setContextCode(Integer.parseInt(line[6]));
subAfterContext.setTimeStamp(Long.parseLong(line[7])); subAfterContext.setTimeStamp(Long.parseLong(line[7]));
subAfterContext.setGroupName(line[8]);
break; break;
default: default:
subAfterContext.setContextCode(Integer.parseInt(line[6])); subAfterContext.setContextCode(Integer.parseInt(line[6]));
subAfterContext.setTimeStamp(Long.parseLong(line[7])); subAfterContext.setTimeStamp(Long.parseLong(line[7]));
subAfterContext.setGroupName(line[8]);
log.warn("Detect new version trace msg of {} type", TraceType.SubAfter.name()); log.warn("Detect new version trace msg of {} type", TraceType.SubAfter.name());
break; break;
} }
resList.add(subAfterContext); resList.add(subAfterContext);
} else if (line[0].equals(TraceType.EndTransaction.name())) { } else if (line[0].equals(TraceType.EndTransaction.name())) {
TraceContext endTransactionContext = new TraceContext(); TraceContext endTransactionContext = initTraceContext();
endTransactionContext.setTraceType(TraceType.EndTransaction); endTransactionContext.setTraceType(TraceType.EndTransaction);
endTransactionContext.setTimeStamp(Long.parseLong(line[1])); endTransactionContext.setTimeStamp(Long.parseLong(line[1]));
endTransactionContext.setRegionId(line[2]); endTransactionContext.setRegionId(line[2]);
@@ -159,4 +161,12 @@ public class MsgTraceDecodeUtil {
} }
return resList; return resList;
} }
private static TraceContext initTraceContext() {
TraceContext traceContext = new TraceContext();
traceContext.setTimeStamp(0L);
traceContext.setCostTime(-1);
traceContext.setRequestId(null);
return traceContext;
}
} }

44
src/main/resources/static/src/messageTrace.js
View File

@@ -129,12 +129,6 @@ module.controller('messageTraceDetailViewDialogController', ['$scope', '$timeout
if (messageTraceGraph.producerNode) { if (messageTraceGraph.producerNode) {
startTime = +messageTraceGraph.producerNode.traceNode.beginTimestamp; startTime = +messageTraceGraph.producerNode.traceNode.beginTimestamp;
endTime = +messageTraceGraph.producerNode.traceNode.endTimestamp; endTime = +messageTraceGraph.producerNode.traceNode.endTimestamp;
} else {
messageTraceGraph.subscriptionNodeList.forEach(subscriptionNode => {
subscriptionNode.consumeNodeList.forEach(consumeNode => {
startTime = Math.min(startTime, consumeNode.beginTimestamp);
})
})
} }
function buildNodeColor(traceNode) { function buildNodeColor(traceNode) {
@@ -203,17 +197,23 @@ module.controller('messageTraceDetailViewDialogController', ['$scope', '$timeout
} }
return `costTime: ${formatCostTimeStr(costTime)}<br/>` return `costTime: ${formatCostTimeStr(costTime)}<br/>`
} }
function buildTimeStamp(timestamp){
if(timestamp < 0){
return 'N/A';
}
return new moment(timestamp).format(TIME_FORMAT_PATTERN);
}
function formatNodeToolTip(params) { function formatNodeToolTip(params) {
let traceNode = params.data.traceData.traceNode; let traceNode = params.data.traceData.traceNode;
return ` return `
${buildCostTimeInfo(traceNode.costTime)} ${buildCostTimeInfo(traceNode.costTime)}
status: ${traceNode.status}<br /> status: ${traceNode.status}<br />
beginTimestamp: ${new moment(traceNode.beginTimestamp).format(TIME_FORMAT_PATTERN)}<br /> ${buildTraceInfo('beginTimestamp', buildTimeStamp(traceNode.beginTimestamp))}
endTimestamp: ${new moment(traceNode.endTimestamp).format(TIME_FORMAT_PATTERN)}<br /> ${buildTraceInfo('endTimestamp', buildTimeStamp(traceNode.endTimestamp))}
clientHost: ${traceNode.clientHost}<br /> clientHost: ${traceNode.clientHost}<br />
storeHost: ${traceNode.storeHost}<br /> storeHost: ${traceNode.storeHost}<br />
retryTimes: ${traceNode.retryTimes}<br /> retryTimes: ${traceNode.retryTimes < 0 ? 'N/A' : traceNode.retryTimes}<br />
${buildTraceInfo('msgType', traceNode.msgType)} ${buildTraceInfo('msgType', traceNode.msgType)}
${buildTraceInfo('transactionId', traceNode.transactionId)} ${buildTraceInfo('transactionId', traceNode.transactionId)}
${buildTraceInfo('transactionState', traceNode.transactionState)} ${buildTraceInfo('transactionState', traceNode.transactionState)}
@@ -221,12 +221,31 @@ module.controller('messageTraceDetailViewDialogController', ['$scope', '$timeout
`; `;
} }
function calcGraphTimestamp(timestamp, relativeTimeStamp, duration, addDuration) {
if (timestamp > 0) {
return timestamp;
}
if (duration < 0) {
return relativeTimeStamp;
}
return addDuration ? relativeTimeStamp + duration : relativeTimeStamp - duration;
}
function addTraceData(traceNode, index) { function addTraceData(traceNode, index) {
if (traceNode.beginTimestamp < 0 && traceNode.endTimestamp < 0) {
return;
}
let beginTimestamp = calcGraphTimestamp(traceNode.beginTimestamp, traceNode.endTimestamp, traceNode.costTime, false);
let endTimestamp = calcGraphTimestamp(traceNode.endTimestamp, traceNode.beginTimestamp, traceNode.costTime, true);
if (endTimestamp === beginTimestamp) {
endTimestamp = beginTimestamp + 1;
}
console.log("beginTimestamp",beginTimestamp,'endTimestamp',endTimestamp);
data.push({ data.push({
value: [ value: [
index, index,
traceNode.beginTimestamp, beginTimestamp,
traceNode.endTimestamp === traceNode.beginTimestamp ? traceNode.beginTimestamp + 1 : traceNode.endTimestamp, endTimestamp,
traceNode.costTime traceNode.costTime
], ],
itemStyle: { itemStyle: {
@@ -239,7 +258,8 @@ module.controller('messageTraceDetailViewDialogController', ['$scope', '$timeout
traceNode: traceNode traceNode: traceNode
} }
}); });
endTime = Math.max(traceNode.endTimestamp, endTime); startTime = Math.min(startTime, beginTimestamp);
endTime = Math.max(endTime, endTimestamp);
} }
messageTraceGraph.subscriptionNodeList.forEach(item => { messageTraceGraph.subscriptionNodeList.forEach(item => {

12
src/main/resources/static/view/pages/messageTrace.html
View File

@@ -273,16 +273,20 @@
</tr> </tr>
<tr ng-repeat="consumeNode in subscriptionNode.consumeNodeList"> <tr ng-repeat="consumeNode in subscriptionNode.consumeNodeList">
<td class="text-center"> <td class="text-center">
{{consumeNode.beginTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss'}} {{consumeNode.beginTimestamp < 0 ? 'N/A' :
(consumeNode.beginTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss')}}
</td> </td>
<td class="text-center"> <td class="text-center">
{{consumeNode.endTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss'}} {{consumeNode.endTimestamp < 0 ? 'N/A' :
(consumeNode.endTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss')}}
</td> </td>
<td class="text-center">{{consumeNode.costTime < 0 ? '--' : <td class="text-center">{{consumeNode.costTime < 0 ? 'N/A' :
((consumeNode.costTime === 0 ? '<1' : consumeNode.costTime) + 'ms')}} ((consumeNode.costTime === 0 ? '<1' : consumeNode.costTime) + 'ms')}}
</td> </td>
<td class="text-center">{{consumeNode.status}}</td> <td class="text-center">{{consumeNode.status}}</td>
<td class="text-center">{{consumeNode.retryTimes}}</td> <td class="text-center">
{{consumeNode.retryTimes < 0 ? 'N/A' : consumeNode.retryTimes}}
</td>
<td class="text-center">{{consumeNode.clientHost}}</td> <td class="text-center">{{consumeNode.clientHost}}</td>
<td class="text-center">{{consumeNode.storeHost}}</td> <td class="text-center">{{consumeNode.storeHost}}</td>
</tr> </tr>