Proxy Support And ConsumerGroup Enhancement (#207)

* Support dashboard v4-v5 switch And query for v5 topic

* Modify tag name

* Support proxy-module And Fix the problem of showing wrong consumerGroup-info

---------

Co-authored-by: yuanziwei <[email protected]>

Author: 1294566108

src/main/java/org/apache/rocketmq/dashboard/config/RMQConfigure.java

@@ -43,6 +43,8 @@ public class RMQConfigure {
     //use rocketmq.namesrv.addr first,if it is empty,than use system proerty or system env
     private volatile String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
 
+    private volatile String proxyAddr;
+
     private volatile String isVIPChannel = System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true");
 
 
@@ -62,6 +64,8 @@ public class RMQConfigure {
 
     private List<String> namesrvAddrs = new ArrayList<>();
 
+    private List<String> proxyAddrs = new ArrayList<>();
+
     public String getAccessKey() {
         return accessKey;
     }
@@ -86,6 +90,25 @@ public List<String> getNamesrvAddrs() {
         return namesrvAddrs;
     }
 
+    public List<String> getProxyAddrs() {
+        return this.proxyAddrs;
+    }
+
+    public void setProxyAddrs(List<String> proxyAddrs) {
+        this.proxyAddrs = proxyAddrs;
+        if (CollectionUtils.isNotEmpty(proxyAddrs)) {
+            this.setProxyAddr(proxyAddrs.get(0));
+        }
+    }
+
+    public String getProxyAddr() {
+        return proxyAddr;
+    }
+
+    public void setProxyAddr(String proxyAddr) {
+        this.proxyAddr = proxyAddr;
+    }
+
     public void setNamesrvAddrs(List<String> namesrvAddrs) {
         this.namesrvAddrs = namesrvAddrs;
         if (CollectionUtils.isNotEmpty(namesrvAddrs)) {

src/main/java/org/apache/rocketmq/dashboard/controller/ConsumerController.java

@@ -47,14 +47,14 @@ public class ConsumerController {
 
     @RequestMapping(value = "/groupList.query")
     @ResponseBody
-    public Object list(@RequestParam(value = "skipSysGroup", required = false) boolean skipSysGroup) {
-        return consumerService.queryGroupList(skipSysGroup);
+    public Object list(@RequestParam(value = "skipSysGroup", required = false) boolean skipSysGroup, String address) {
+        return consumerService.queryGroupList(skipSysGroup, address);
     }
 
     @RequestMapping(value = "/group.query")
     @ResponseBody
-    public Object groupQuery(@RequestParam String consumerGroup) {
-        return consumerService.queryGroup(consumerGroup);
+    public Object groupQuery(@RequestParam String consumerGroup, String address) {
+        return consumerService.queryGroup(consumerGroup, address);
     }
 
     @RequestMapping(value = "/resetOffset.do", method = {RequestMethod.POST})
@@ -99,14 +99,14 @@ public Object fetchBrokerNameList(@RequestParam String consumerGroup) {
 
     @RequestMapping(value = "/queryTopicByConsumer.query")
     @ResponseBody
-    public Object queryConsumerByTopic(@RequestParam String consumerGroup) {
-        return consumerService.queryConsumeStatsListByGroupName(consumerGroup);
+    public Object queryConsumerByTopic(@RequestParam String consumerGroup, String address) {
+        return consumerService.queryConsumeStatsListByGroupName(consumerGroup, address);
     }
 
     @RequestMapping(value = "/consumerConnection.query")
     @ResponseBody
-    public Object consumerConnection(@RequestParam(required = false) String consumerGroup) {
-        ConsumerConnection consumerConnection = consumerService.getConsumerConnection(consumerGroup);
+    public Object consumerConnection(@RequestParam(required = false) String consumerGroup, String address) {
+        ConsumerConnection consumerConnection = consumerService.getConsumerConnection(consumerGroup, address);
         consumerConnection.setConnectionSet(ConnectionInfo.buildConnectionInfoHashSet(consumerConnection.getConnectionSet()));
         return consumerConnection;
     }

src/main/java/org/apache/rocketmq/dashboard/controller/ProxyController.java

@@ -0,0 +1,54 @@
+/*
+ * 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.rocketmq.dashboard.controller;
+
+import org.apache.rocketmq.dashboard.permisssion.Permission;
+import org.apache.rocketmq.dashboard.service.ProxyService;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+import javax.annotation.Resource;
+
+@Controller
+@RequestMapping("/proxy")
+@Permission
+public class ProxyController {
+    @Resource
+    private ProxyService proxyService;
+    @RequestMapping(value = "/homePage.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object homePage() {
+        return proxyService.getProxyHomePage();
+    }
+
+    @RequestMapping(value = "/addProxyAddr.do", method = RequestMethod.POST)
+    @ResponseBody
+    public Object addProxyAddr(@RequestParam String newProxyAddr) {
+        proxyService.addProxyAddrList(newProxyAddr);
+        return true;
+    }
+
+    @RequestMapping(value = "/updateProxyAddr.do", method = RequestMethod.POST)
+    @ResponseBody
+    public Object updateProxyAddr(@RequestParam String proxyAddr) {
+        proxyService.updateProxyAddrList(proxyAddr);
+        return true;
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/model/GroupConsumeInfo.java

@@ -19,12 +19,15 @@
 import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
 import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
 
+import java.util.List;
+
 public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
     private String group;
     private String version;
     private int count;
     private ConsumeType consumeType;
     private MessageModel messageModel;
+    private List<String> address;
     private int consumeTps;
     private long diffTotal = -1;
     private String subGroupType = "NORMAL";
@@ -70,6 +73,22 @@ public void setDiffTotal(long diffTotal) {
         this.diffTotal = diffTotal;
     }
 
+    public List<String> getAddress() {
+        return address;
+    }
+
+    public void setAddress(List<String> address) {
+        this.address = address;
+    }
+
+    public String getSubGroupType() {
+        return subGroupType;
+    }
+
+    public void setSubGroupType(String subGroupType) {
+        this.subGroupType = subGroupType;
+    }
+
     @Override
     public int compareTo(GroupConsumeInfo o) {
         if (this.count != o.count) {
@@ -93,12 +112,4 @@ public String getVersion() {
     public void setVersion(String version) {
         this.version = version;
     }
-
-    public String getSubGroupType() {
-        return subGroupType;
-    }
-
-    public void setSubGroupType(String subGroupType) {
-        this.subGroupType = subGroupType;
-    }
 }

src/main/java/org/apache/rocketmq/dashboard/service/ConsumerService.java

@@ -31,12 +31,12 @@
 import java.util.Set;
 
 public interface ConsumerService {
-    List<GroupConsumeInfo> queryGroupList(boolean skipSysGroup);
+    List<GroupConsumeInfo> queryGroupList(boolean skipSysGroup,String address);
 
-    GroupConsumeInfo queryGroup(String consumerGroup);
+    GroupConsumeInfo queryGroup(String consumerGroup, String address);
 
 
-    List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName);
+    List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName, String address);
 
     List<TopicConsumerInfo> queryConsumeStatsList(String topic, String groupName);
 
@@ -52,7 +52,7 @@ public interface ConsumerService {
 
     Set<String> fetchBrokerNameSetBySubscriptionGroup(String group);
 
-    ConsumerConnection getConsumerConnection(String consumerGroup);
+    ConsumerConnection getConsumerConnection(String consumerGroup, String address);
 
     ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack);
 }

src/main/java/org/apache/rocketmq/dashboard/service/ProxyService.java

@@ -0,0 +1,28 @@
+/*
+ * 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.rocketmq.dashboard.service;
+
+import java.util.Map;
+
+public interface ProxyService {
+
+    void addProxyAddrList(String proxyAddr);
+
+    void updateProxyAddrList(String proxyAddr);
+
+    Map<String, Object>  getProxyHomePage();
+}

src/main/java/org/apache/rocketmq/dashboard/service/client/MQAdminExtImpl.java

@@ -627,7 +627,7 @@ public ConsumeStats examineConsumeStats(String brokerAddr, String consumerGroup,
             long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
             RemotingConnectException, MQBrokerException {
         // TODO Auto-generated method stub
-        throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStats'");
+        return MQAdminInstance.threadLocalMQAdminExt().examineConsumeStats(brokerAddr, consumerGroup, topicName, timeoutMillis);
     }
 
     @Override
@@ -639,8 +639,7 @@ public AdminToolResult<ConsumeStats> examineConsumeStatsConcurrent(String consum
     @Override
     public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup, String brokerAddr)
             throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
-        // TODO Auto-generated method stub
-        throw new UnsupportedOperationException("Unimplemented method 'examineConsumerConnectionInfo'");
+        return MQAdminInstance.threadLocalMQAdminExt().examineConsumerConnectionInfo(consumerGroup, brokerAddr);
     }
 
     @Override

src/main/java/org/apache/rocketmq/dashboard/service/client/ProxyAdmin.java

@@ -0,0 +1,28 @@
+/*
+ * 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.rocketmq.dashboard.service.client;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+
+public interface ProxyAdmin {
+
+    ConsumerConnection examineConsumerConnectionInfo(String addr, String consumerGroup) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException;
+}

src/main/java/org/apache/rocketmq/dashboard/service/client/ProxyAdminImpl.java

@@ -0,0 +1,60 @@
+/*
+ * 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.rocketmq.dashboard.service.client;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import static org.apache.rocketmq.remoting.protocol.RequestCode.GET_CONSUMER_CONNECTION_LIST;
+
+@Slf4j
+@Service
+public class ProxyAdminImpl implements ProxyAdmin {
+    @Autowired
+    private GenericObjectPool<MQAdminExt> mqAdminExtPool;
+
+    @Override
+    public ConsumerConnection examineConsumerConnectionInfo(String addr, String consumerGroup) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
+        try {
+            MQAdminInstance.createMQAdmin(mqAdminExtPool);
+            RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
+            GetConsumerConnectionListRequestHeader requestHeader = new GetConsumerConnectionListRequestHeader();
+            requestHeader.setConsumerGroup(consumerGroup);
+            RemotingCommand request = RemotingCommand.createRequestCommand(GET_CONSUMER_CONNECTION_LIST, requestHeader);
+            RemotingCommand response = remotingClient.invokeSync(addr, request, 3000);
+            switch (response.getCode()) {
+                case 0:
+                    return ConsumerConnection.decode(response.getBody(), ConsumerConnection.class);
+                default:
+                    throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
+            }
+        } finally {
+            MQAdminInstance.returnMQAdmin(mqAdminExtPool);
+        }
+    }
+}