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

[ISSUE #19]Message track query enhancement (#21)

Browse Source 
* [ISSUE #19]Message track query enhancement

* traceTopic can be null and system Topic is used by default

* add unit test

* select messageTrack topic in messageTrace page

Co-authored-by: zhangjidi <zhangjidi@cmss.chinamobile.com>
This commit is contained in:
zhangjidi2016
2021-09-23 21:01:32 +08:00
committed by GitHub
parent dc67c660ff
commit 58336d951a
16 changed files with 97 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/main/java/org/apache/rocketmq/dashboard/config/RMQConfigure.java
View File

@@ -18,7 +18,6 @@ package org.apache.rocketmq.dashboard.config;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -48,8 +47,6 @@ public class RMQConfigure {
private boolean enableDashBoardCollect;
private String msgTrackTopicName;
private boolean loginRequired = false;
private String accessKey;
@@ -123,17 +120,6 @@ public class RMQConfigure {
this.enableDashBoardCollect = Boolean.valueOf(enableDashBoardCollect);
}
public String getMsgTrackTopicNameOrDefault() {
if (StringUtils.isEmpty(msgTrackTopicName)) {
return TopicValidator.RMQ_SYS_TRACE_TOPIC;
}
return msgTrackTopicName;
}
public void setMsgTrackTopicName(String msgTrackTopicName) {
this.msgTrackTopicName = msgTrackTopicName;
}
public boolean isLoginRequired() {
return loginRequired;
}

5
src/main/java/org/apache/rocketmq/dashboard/controller/MessageTraceController.java
View File

@@ -64,7 +64,8 @@ public class MessageTraceController {
@RequestMapping(value = "/viewMessageTraceGraph.query", method = RequestMethod.GET)
@ResponseBody
public MessageTraceGraph viewMessageTraceGraph(@RequestParam String msgId) {
return messageTraceService.queryMessageTraceGraph(msgId);
public MessageTraceGraph viewMessageTraceGraph(@RequestParam String msgId,
@RequestParam(required = false) String traceTopic) {
return messageTraceService.queryMessageTraceGraph(msgId, traceTopic);
}
}

9
src/main/java/org/apache/rocketmq/dashboard/controller/TopicController.java
View File

@@ -51,12 +51,9 @@ public class TopicController {
@RequestMapping(value = "/list.query", method = RequestMethod.GET)
@ResponseBody
public Object list(@RequestParam(value = "skipSysProcess", required = false) String skipSysProcess) {
boolean flag = false;
if ("true".equals(skipSysProcess)) {
flag = true;
}
return topicService.fetchAllTopicList(flag);
public Object list(@RequestParam(value = "skipSysProcess", required = false) boolean skipSysProcess,
@RequestParam(value = "skipRetryAndDlq", required = false) boolean skipRetryAndDlq) {
return topicService.fetchAllTopicList(skipSysProcess, skipRetryAndDlq);
}
@RequestMapping(value = "/stats.query", method = RequestMethod.GET)

2
src/main/java/org/apache/rocketmq/dashboard/service/MessageTraceService.java
View File

@@ -27,5 +27,5 @@ public interface MessageTraceService {
List<MessageTraceView> queryMessageTraceByTopicAndKey(final String topic, final String key);
MessageTraceGraph queryMessageTraceGraph(final String key);
MessageTraceGraph queryMessageTraceGraph(final String key, final String traceTopic);
}

2
src/main/java/org/apache/rocketmq/dashboard/service/TopicService.java
View File

@@ -29,7 +29,7 @@ import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
import java.util.List;
public interface TopicService {
TopicList fetchAllTopicList(boolean skipSysProcess);
TopicList fetchAllTopicList(boolean skipSysProcess, boolean skipRetryAndDlq);
TopicStatsTable stats(String topic);

10
src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageTraceServiceImpl.java
View File

@@ -33,6 +33,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.trace.TraceType;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.model.MessageTraceView;
import org.apache.rocketmq.dashboard.model.trace.ProducerNode;
@@ -65,7 +66,7 @@ public class MessageTraceServiceImpl implements MessageTraceService {
@Override
public List<MessageTraceView> queryMessageTraceKey(String key) {
String queryTopic = configure.getMsgTrackTopicNameOrDefault();
String queryTopic = TopicValidator.RMQ_SYS_TRACE_TOPIC;
logger.info("query data topic name is:{}", queryTopic);
return queryMessageTraceByTopicAndKey(queryTopic, key);
}
@@ -86,8 +87,11 @@ public class MessageTraceServiceImpl implements MessageTraceService {
}
@Override
public MessageTraceGraph queryMessageTraceGraph(String key) {
List<MessageTraceView> messageTraceViews = queryMessageTraceKey(key);
public MessageTraceGraph queryMessageTraceGraph(String key, String topic) {
if (StringUtils.isEmpty(topic)) {
topic = TopicValidator.RMQ_SYS_TRACE_TOPIC;
}
List<MessageTraceView> messageTraceViews = queryMessageTraceByTopicAndKey(topic, key);
return buildMessageTraceGraph(messageTraceViews);
}

32
src/main/java/org/apache/rocketmq/dashboard/service/impl/TopicServiceImpl.java
View File

@@ -20,6 +20,7 @@ package org.apache.rocketmq.dashboard.service.impl;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
@@ -36,6 +37,7 @@ import org.apache.rocketmq.common.protocol.body.GroupList;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
@@ -61,23 +63,23 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
private RMQConfigure configure;
@Override
public TopicList fetchAllTopicList(boolean skipSysProcess) {
public TopicList fetchAllTopicList(boolean skipSysProcess, boolean skipRetryAndDlq) {
try {
TopicList allTopics = mqAdminExt.fetchAllTopicList();
if (skipSysProcess) {
return allTopics;
}
TopicList sysTopics = getSystemTopicList();
Set<String> topics = new HashSet<>();
for (String topic : allTopics.getTopicList()) {
if (sysTopics.getTopicList().contains(topic)) {
topics.add(String.format("%s%s", "%SYS%", topic));
} else {
topics.add(topic);
}
}
Set<String> topics =
allTopics.getTopicList().stream().map(topic -> {
if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
topic = String.format("%s%s", "%SYS%", topic);
}
return topic;
}).filter(topic -> {
if (skipRetryAndDlq) {
return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
|| topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
}
return true;
}).collect(Collectors.toSet());
allTopics.getTopicList().clear();
allTopics.getTopicList().addAll(topics);
return allTopics;
@@ -209,7 +211,7 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
}
public DefaultMQProducer buildDefaultMQProducer(String producerGroup, RPCHook rpcHook, boolean traceEnabled) {
DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup, rpcHook, traceEnabled, configure.getMsgTrackTopicNameOrDefault());
DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup, rpcHook, traceEnabled, TopicValidator.RMQ_SYS_TRACE_TOPIC);
defaultMQProducer.setUseTLS(configure.isUseTLS());
return defaultMQProducer;
}

5
src/main/resources/static/src/i18n/en.js
View File

@@ -105,5 +105,8 @@ var en = {
"NO_DATA":"Don't have ",
"SYSTEM":"SYSTEM",
"WELCOME":"Hi, welcome using RocketMQ Dashboard",
"ENABLE_MESSAGE_TRACE":"Enable Message Trace"
"ENABLE_MESSAGE_TRACE":"Enable Message Trace",
"MESSAGE_TRACE_DETAIL":"Message Trace Detail",
"TRACE_TOPIC":"TraceTopic",
"SELECT_TRACE_TOPIC":"selectTraceTopic"
}

5
src/main/resources/static/src/i18n/zh.js
View File

@@ -106,5 +106,8 @@ var zh = {
"NO_DATA":"不存在 ",
"SYSTEM":"系统",
"WELCOME":"您好，欢迎使用RocketMQ仪表盘",
"ENABLE_MESSAGE_TRACE":"开启消息轨迹"
"ENABLE_MESSAGE_TRACE":"开启消息轨迹",
"MESSAGE_TRACE_DETAIL":"消息轨迹详情",
"TRACE_TOPIC":"消息轨迹主题",
"SELECT_TRACE_TOPIC":"选择消息轨迹主题"
}

2
src/main/resources/static/src/message.js
View File

@@ -29,7 +29,7 @@ module.controller('messageController', ['$scope', 'ngDialog', '$http', 'Notifica
method: "GET",
url: "topic/list.query",
params: {
skipSysProcess: 'true'
skipSysProcess: true
}
}).success(function (resp) {
if (resp.status == 0) {

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

@@ -26,9 +26,13 @@ const TIME_FORMAT_PATTERN = "YYYY-MM-DD HH:mm:ss.SSS";
const DEFAULT_DISPLAY_DURATION = 10 * 1000
// transactionTraceNode do not have costTime, assume it cost 50ms
const TRANSACTION_CHECK_COST_TIME = 50;
const RETRY_GROUP_TOPIC_PREFIX = "%RETRY%";
const DLQ_GROUP_TOPIC_PREFIX = "%DLQ%";
module.controller('messageTraceController', ['$scope', '$routeParams', 'ngDialog', '$http', 'Notification', function ($scope, $routeParams, ngDialog, $http, Notification) {
$scope.allTopicList = [];
$scope.selectedTopic = [];
$scope.allTraceTopicList = [];
$scope.selectedTraceTopic = [];
$scope.key = "";
$scope.messageId = $routeParams.messageId;
$scope.queryMessageByTopicAndKeyResult = [];
@@ -39,16 +43,25 @@ module.controller('messageTraceController', ['$scope', '$routeParams', 'ngDialog
method: "GET",
url: "topic/list.query",
params: {
skipSysProcess: "true"
skipSysProcess: true
}
}).success(function (resp) {
if (resp.status == 0) {
$scope.allTopicList = resp.data.topicList.sort();
console.log($scope.allTopicList);
console.log($scope.allTopicList)
for (const topic of $scope.allTopicList) {
if (topic.startsWith(RETRY_GROUP_TOPIC_PREFIX)
|| topic.startsWith(DLQ_GROUP_TOPIC_PREFIX)) {
continue;
}
$scope.allTraceTopicList.push(topic);
}
console.log($scope.allTraceTopicList)
} else {
Notification.error({message: resp.errMsg, delay: 2000});
}
});
$scope.timepickerBegin = moment().subtract(1, 'hour').format('YYYY-MM-DD HH:mm');
$scope.timepickerEnd = moment().add(1, 'hour').format('YYYY-MM-DD HH:mm');
$scope.timepickerOptions = {format: 'YYYY-MM-DD HH:mm', showClear: true};
@@ -99,7 +112,7 @@ module.controller('messageTraceController', ['$scope', '$routeParams', 'ngDialog
url: "messageTrace/viewMessageTraceGraph.query",
params: {
msgId: messageId,
topic: topic
traceTopic: topic
}
}).success(function (resp) {
if (resp.status == 0) {

2
src/main/resources/static/src/producer.js
View File

@@ -23,7 +23,7 @@ module.controller('producerController', ['$scope', '$http','Notification',functi
method: "GET",
url: "topic/list.query",
params:{
skipSysProcess:"true"
skipSysProcess: true
}
}).success(function (resp) {
if(resp.status ==0){

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

@@ -15,6 +15,20 @@
~ limitations under the License.
-->
<div class="container-fluid" id="deployHistoryList">
<div class="modal-header">
<div class="row">
<label style="color: #000000">{{ 'TRACE_TOPIC' | translate }}:</label>
<div style="display: inline-block; min-width: 300px">
<select name="mySelect" chosen
ng-model="selectedTraceTopic"
ng-options="item for item in allTraceTopicList"
required>
<option value=""></option>
</select>
</div>
<div style="display: inline-block; color: #BDBDBD">(if no select, it will use RMQ_SYS_TRACE_TOPIC)</div>
</div>
</div>
<div class="modal-body">
<div ng-cloak="" class="tabsdemoDynamicHeight">
<md-content>
@@ -64,7 +78,7 @@
</td>
<td class="text-center">
<button class="btn btn-raised btn-sm btn-primary" type="button"
ng-click="queryMessageTraceByMessageId(item.msgId,item.topic)">Message Trace Detail
ng-click="queryMessageTraceByMessageId(item.msgId, selectedTraceTopic)">{{ 'MESSAGE_TRACE_DETAIL' | translate }}
</button>
</td>
</tr>
@@ -114,7 +128,7 @@
</td>
<td class="text-center">
<button class="btn btn-raised btn-sm btn-primary" type="button"
ng-click="queryMessageTraceByMessageId(item.msgId,item.topic)">Message Trace Detail
ng-click="queryMessageTraceByMessageId(item.msgId, selectedTraceTopic)">{{ 'MESSAGE_TRACE_DETAIL' | translate }}
</button>
</td>
</tr>
@@ -274,14 +288,14 @@
<tr ng-repeat="consumeNode in subscriptionNode.consumeNodeList">
<td class="text-center">
{{consumeNode.beginTimestamp < 0 ? 'N/A' :
(consumeNode.beginTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss')}}
(consumeNode.beginTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss')}}
</td>
<td class="text-center">
{{consumeNode.endTimestamp < 0 ? 'N/A' :
(consumeNode.endTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss')}}
(consumeNode.endTimestamp | date:'yyyy-MM-dd HH:mm:ss.sss')}}
</td>
<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 class="text-center">{{consumeNode.status}}</td>
<td class="text-center">
@@ -302,10 +316,12 @@
</div>
</div>
</md-content>
<div class="ngdialog-buttons">
<button type="button" class="ngdialog-button ngdialog-button-secondary"
ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
</button>
<div class="modal-footer">
<div class="ngdialog-buttons">
<button type="button" class="ngdialog-button ngdialog-button-secondary"
ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
</button>
</div>
</div>
</script>

3
src/test/java/org/apache/rocketmq/dashboard/config/RMQConfigureTest.java
View File

@@ -18,7 +18,6 @@
package org.apache.rocketmq.dashboard.config;
import java.io.File;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.boot.web.server.ErrorPage;
@@ -38,7 +37,6 @@ public class RMQConfigureTest {
rmqConfigure.setIsVIPChannel("true");
rmqConfigure.setUseTLS(true);
rmqConfigure.setLoginRequired(true);
rmqConfigure.setMsgTrackTopicName(null);
rmqConfigure.setNamesrvAddr("127.0.0.1:9876");
rmqConfigure.setTimeoutMillis(3000L);
}
@@ -55,7 +53,6 @@ public class RMQConfigureTest {
Assert.assertEquals(rmqConfigure.getIsVIPChannel(), "true");
Assert.assertTrue(rmqConfigure.isEnableDashBoardCollect());
Assert.assertTrue(rmqConfigure.isLoginRequired());
Assert.assertEquals(rmqConfigure.getMsgTrackTopicNameOrDefault(), TopicValidator.RMQ_SYS_TRACE_TOPIC);
Assert.assertEquals(rmqConfigure.getNamesrvAddr(), "127.0.0.1:9876");
Assert.assertEquals(rmqConfigure.getTimeoutMillis().longValue(), 3000L);
ErrorPageRegistrar registrar = rmqConfigure.errorPageRegistrar();

2
src/test/java/org/apache/rocketmq/dashboard/controller/MessageTraceControllerTest.java
View File

@@ -22,7 +22,6 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.trace.TraceType;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.dashboard.service.impl.MessageServiceImpl;
import org.apache.rocketmq.dashboard.service.impl.MessageTraceServiceImpl;
import org.apache.rocketmq.dashboard.util.MockObjectUtil;
@@ -57,7 +56,6 @@ public class MessageTraceControllerTest extends BaseControllerTest {
@Before
public void init() throws MQClientException, InterruptedException {
super.mockRmqConfigure();
when(configure.getMsgTrackTopicNameOrDefault()).thenReturn(TopicValidator.RMQ_SYS_TRACE_TOPIC);
List<MessageExt> messageList = new ArrayList<>(2);
MessageExt messageExt = MockObjectUtil.createMessageExt();
messageExt.setBody(MockObjectUtil.createTraceData().getBytes());

16
src/test/java/org/apache/rocketmq/dashboard/controller/TopicControllerTest.java
View File

@@ -44,6 +44,8 @@ import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
import org.apache.rocketmq.dashboard.service.impl.ConsumerServiceImpl;
import org.apache.rocketmq.dashboard.service.impl.TopicServiceImpl;
import org.apache.rocketmq.dashboard.util.MockObjectUtil;
import org.apache.rocketmq.remoting.RPCHook;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.InjectMocks;
@@ -91,6 +93,8 @@ public class TopicControllerTest extends BaseControllerTest {
topicSet.add("common_topic2");
topicSet.add("system_topic1");
topicSet.add("system_topic2");
topicSet.add("%DLQ%topic");
topicSet.add("%RETRY%topic");
topicList.setTopicList(topicSet);
when(mqAdminExt.fetchAllTopicList()).thenReturn(topicList);
// mock system topics
@@ -117,10 +121,18 @@ public class TopicControllerTest extends BaseControllerTest {
requestBuilder = MockMvcRequestBuilders.get(url);
requestBuilder.param("skipSysProcess", String.valueOf(true));
perform = mockMvc.perform(requestBuilder);
perform.andExpect(status().isOk())
.andExpect(jsonPath("$.data.topicList", hasSize(6)));
// 2、list all topic filter DLQ and Retry topic
requestBuilder = MockMvcRequestBuilders.get(url);
requestBuilder.param("skipSysProcess", String.valueOf(false));
requestBuilder.param("skipRetryAndDlq", String.valueOf(true));
perform = mockMvc.perform(requestBuilder);
perform.andExpect(status().isOk())
.andExpect(jsonPath("$.data.topicList", hasSize(4)));
// 2、filter system topic
// 3、filter system topic
requestBuilder = MockMvcRequestBuilders.get(url);
perform = mockMvc.perform(requestBuilder);
perform.andExpect(status().isOk())
@@ -249,7 +261,7 @@ public class TopicControllerTest extends BaseControllerTest {
when(producer.send(any(Message.class))).thenReturn(result);
doReturn(producer).when(topicService).buildDefaultMQProducer(anyString(), any(), anyBoolean());
}
Assert.assertNotNull(topicService.buildDefaultMQProducer("group_test", mock(RPCHook.class)));
SendTopicMessageRequest request = new SendTopicMessageRequest();
request.setTopic(topicName);
request.setMessageBody("hello world");