add watchdog

.gitignore

@@ -32,6 +32,4 @@ Thumbs.db
 # *.zip
 # *.tar
 # *.tar.gz
-catalina.base_IS_UNDEFINED/
-
-qconfig_test
+catalina.base_IS_UNDEFINED/

pom.xml

@@ -140,6 +140,11 @@
                 <artifactId>qmq-delay-server</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>${project.groupId}</groupId>
+                <artifactId>qmq-watchdog</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>${project.groupId}</groupId>
                 <artifactId>qmq-tools</artifactId>

qmq-dist/bin/watchdog-env.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+JAVA_HOME=""
+JAVA_OPTS="-Xms512m -Xmx512m"
+

qmq-dist/bin/watchdog.cmd

@@ -0,0 +1,11 @@
+@echo off
+
+setlocal
+call "%~dp0base.cmd"
+
+set MAIN=qunar.tc.qmq.task.Bootstrap
+
+echo on
+"%JAVA%" -cp "%CLASSPATH%" "%MAIN%" %*
+
+endlocal

qmq-dist/bin/watchdog.sh

@@ -0,0 +1,67 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+QMQ_BIN="${BASH_SOURCE-$0}"
+QMQ_BIN="$(dirname "$QMQ_BIN")"
+QMQ_BIN_DIR="$(cd "$QMQ_BIN"; pwd)"
+QMQ_META_MAIN="qunar.tc.qmq.task.Bootstrap"
+
+. "$QMQ_BIN_DIR/base.sh"
+. "$QMQ_BIN_DIR/watchdog-env.sh"
+
+if [[ "$JAVA_HOME" != "" ]]; then
+  JAVA="$JAVA_HOME/bin/java"
+else
+  JAVA=java
+fi
+
+JAVA_OPTS="$JAVA_OPTS -DQMQ_LOG_DIR=$QMQ_LOG_DIR"
+QMQ_PID_FILE="$QMQ_PID_DIR/watchdog.pid"
+QMQ_DAEMON_OUT="$QMQ_LOG_DIR/watchdog.out"
+
+CMD=${1:-}
+case ${CMD} in
+start)
+    echo  -n "Starting qmq watchdog ... "
+    if [[ -f "$QMQ_PID_FILE" ]]; then
+      if kill -0 `cat "$QMQ_PID_FILE"` > /dev/null 2>&1; then
+         echo already running as process `cat "$QMQ_PID_FILE"`.
+         exit 0
+      fi
+    fi
+    nohup "$JAVA" -cp "$CLASSPATH" ${JAVA_OPTS} ${QMQ_META_MAIN} > "$QMQ_DAEMON_OUT" 2>&1 < /dev/null &
+    if [[ $? -eq 0 ]]
+    then
+      /bin/echo -n $! > "$QMQ_PID_FILE"
+      if [[ $? -eq 0 ]];
+      then
+        sleep 1
+        echo STARTED
+      else
+        echo FAILED TO WRITE PID
+        exit 1
+      fi
+    else
+      echo SERVER DID NOT START
+      exit 1
+    fi
+    ;;
+start-foreground)
+    ZOO_CMD=(exec "$JAVA")
+    "${ZOO_CMD[@]}" -cp "$CLASSPATH" ${JAVA_OPTS} ${QMQ_META_MAIN}
+    ;;
+stop)
+    echo -n "Stopping qmq watchdog ... "
+    if [[ ! -f "$QMQ_PID_FILE" ]]
+    then
+      echo "no watchdog to stop (could not find file $QMQ_PID_FILE)"
+    else
+      kill -9 $(cat "$QMQ_PID_FILE")
+      rm "$QMQ_PID_FILE"
+      echo STOPPED
+    fi
+    exit 0
+    ;;
+*)
+    echo "Usage: $0 {start|start-foreground|stop}" >&2
+esac

qmq-dist/conf/watchdog.properties

@@ -0,0 +1,8 @@
+# (可选)每次扫描客户db如果超过3分钟则退出，等待下次扫描
+sendMessageTaskExecuteTimeout=180000
+#(可选) 多久刷新一次db里的客户端db记录状态
+refreshInterval=180000
+#(可选) 客户端db里的消息多久之前的认为可以重发
+checkInterval=60000
+#(可选) watchdog可以启动多个，与客户端db里的room对应
+namespace=default

qmq-dist/pom.xml

@@ -28,6 +28,10 @@
             <groupId>${project.groupId}</groupId>
             <artifactId>qmq-delay-server</artifactId>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>qmq-watchdog</artifactId>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>qmq-tools</artifactId>

qmq-dist/sql/init.sql

@@ -91,6 +91,26 @@ CREATE TABLE `broker`
   UNIQUE KEY uniq_broker (`hostname`, `serve_port`),
   UNIQUE KEY uniq_group_role (`group_name`, `role`),
   KEY idx_broker_group (`group_name`)
-) ENGINE InnoDB
-  DEFAULT CHARSET = utf8mb4
-  COMMENT 'broker列表';
+) ENGINE InnoDB DEFAULT CHARSET = utf8mb4 COMMENT 'broker列表';
+
+CREATE TABLE leader_election (
+  `id` int(10) unsigned not null auto_increment comment '主键id',
+  `name` varchar(128) not null default '' comment '名称',
+  `node` varchar(128) NOT NULL default '' comment '节点',
+  `last_seen_active` bigint(20) NOT NULL DEFAULT 0 comment '最后更新时间',
+  PRIMARY KEY (id),
+  unique key uniq_idx_name(name)
+) ENGINE=InnoDB default charset=utf8mb4 comment 'leader选举';
+
+CREATE TABLE datasource_config (
+  `id` int(10) unsigned not null auto_increment comment '主键id',
+  `url` varchar(128) not null default '' comment 'jdbc url',
+  `user_name` varchar(100) NOT NULL default '' comment 'db username',
+  `password` varchar(100) NOT NULL DEFAULT '' comment 'db password',
+  `status` TINYINT NOT NULL DEFAULT 0 COMMENT 'db状态',
+  `room`           VARCHAR(20)      NOT NULL DEFAULT '' COMMENT '机房',
+  `create_time` TIMESTAMP NOT NULL DEFAULT '2018-01-01 01:01:01' COMMENT '创建时间',
+  `update_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
+  PRIMARY KEY (id),
+  unique key uniq_idx_name(name)
+) ENGINE=InnoDB default charset=utf8mb4 comment '客户端db配置表';

qmq-metaserver/src/main/java/qunar/tc/qmq/meta/management/RegisterClientDbAction.java

@@ -0,0 +1,43 @@
+package qunar.tc.qmq.meta.management;
+
+import qunar.tc.qmq.meta.model.ClientDbInfo;
+import qunar.tc.qmq.meta.store.ClientDbConfigurationStore;
+
+import javax.servlet.http.HttpServletRequest;
+import java.util.Optional;
+
+public class RegisterClientDbAction implements MetaManagementAction {
+
+    private final ClientDbConfigurationStore store;
+
+    public RegisterClientDbAction(ClientDbConfigurationStore store) {
+        this.store = store;
+    }
+
+    @Override
+    public Object handleAction(HttpServletRequest req) {
+        String type = req.getParameter("type");
+        String host = req.getParameter("host");
+        int port = Integer.valueOf(req.getParameter("port"));
+        String userName = req.getParameter("userName");
+        String password = req.getParameter("password");
+
+        String room = req.getParameter("room");
+        if (room == null || room.trim().length() == 0) {
+            room = "default";
+        }
+
+        String url = type + "://" + host + ":" + port;
+        ClientDbInfo clientDbInfo = new ClientDbInfo();
+        clientDbInfo.setUrl(url);
+        clientDbInfo.setUserName(userName);
+        clientDbInfo.setPassword(password);
+        clientDbInfo.setRoom(room);
+        try {
+            store.insertDb(clientDbInfo);
+            return Optional.of("register success");
+        } catch (Exception e) {
+            return Optional.of("register failed " + e.getMessage());
+        }
+    }
+}

qmq-metaserver/src/main/java/qunar/tc/qmq/meta/model/ClientDbInfo.java

@@ -0,0 +1,44 @@
+package qunar.tc.qmq.meta.model;
+
+public class ClientDbInfo {
+    private String url;
+
+    private String userName;
+
+    private String password;
+
+    private String room;
+
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getUserName() {
+        return userName;
+    }
+
+    public void setUserName(String userName) {
+        this.userName = userName;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getRoom() {
+        return room;
+    }
+
+    public void setRoom(String room) {
+        this.room = room;
+    }
+}

qmq-metaserver/src/main/java/qunar/tc/qmq/meta/startup/ServerWrapper.java

@@ -31,9 +31,11 @@
 import qunar.tc.qmq.meta.route.impl.DefaultSubjectRouter;
 import qunar.tc.qmq.meta.route.impl.DelayRouter;
 import qunar.tc.qmq.meta.store.BrokerStore;
+import qunar.tc.qmq.meta.store.ClientDbConfigurationStore;
 import qunar.tc.qmq.meta.store.JdbcTemplateHolder;
 import qunar.tc.qmq.meta.store.Store;
 import qunar.tc.qmq.meta.store.impl.BrokerStoreImpl;
+import qunar.tc.qmq.meta.store.impl.ClientDbConfigurationStoreImpl;
 import qunar.tc.qmq.meta.store.impl.DatabaseStore;
 import qunar.tc.qmq.netty.DefaultConnectionEventHandler;
 import qunar.tc.qmq.netty.NettyServer;
@@ -81,6 +83,8 @@ public void start(ServletContext context) {
         metaNettyServer.registerProcessor(CommandCode.BROKER_ACQUIRE_META, brokerAcquireMetaProcessor);
         metaNettyServer.start();
 
+        ClientDbConfigurationStore clientDbConfigurationStore = new ClientDbConfigurationStoreImpl();
+
         final MetaManagementActionSupplier actions = MetaManagementActionSupplier.getInstance();
         actions.register("AddBroker", new TokenVerificationAction(new AddBrokerAction(brokerStore)));
         actions.register("ReplaceBroker", new TokenVerificationAction(new ReplaceBrokerAction(brokerStore)));
@@ -91,6 +95,7 @@ public void start(ServletContext context) {
         actions.register("RemoveSubjectBrokerGroup", new TokenVerificationAction(new RemoveSubjectBrokerGroupAction(store, cachedMetaInfoManager)));
         actions.register("AddNewSubject", new TokenVerificationAction(new AddNewSubjectAction(store)));
         actions.register("ExtendSubjectRoute", new TokenVerificationAction(new ExtendSubjectRouteAction(store, cachedMetaInfoManager)));
+        actions.register("AddDb", new TokenVerificationAction(new RegisterClientDbAction(clientDbConfigurationStore)));
 
         resources.add(cachedMetaInfoManager);
         resources.add(metaNettyServer);

qmq-metaserver/src/main/java/qunar/tc/qmq/meta/store/ClientDbConfigurationStore.java

@@ -0,0 +1,7 @@
+package qunar.tc.qmq.meta.store;
+
+import qunar.tc.qmq.meta.model.ClientDbInfo;
+
+public interface ClientDbConfigurationStore {
+    void insertDb(ClientDbInfo clientDbInfo);
+}