From bf89974f7c7933661696013220a99fc9c3cb331c Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 27 Aug 2019 16:11:20 +0800 Subject: [PATCH] fixed issue #2099 , support remote config poll --- canal-admin/canal-admin-server/pom.xml | 35 ++- .../src/main/assembly/release.xml | 41 +++ .../src/main/bin/restart.sh | 5 + .../src/main/bin/startup.bat | 22 ++ .../src/main/bin/startup.sh | 84 ++++++ .../canal-admin-server/src/main/bin/stop.sh | 65 +++++ .../otter/canal/admin/config/WebConfig.java | 5 +- .../controller/CanalConfigController.java | 19 +- .../controller/PollConfigController.java | 155 ++++++++++ .../otter/canal/admin/model/CanalConfig.java | 10 + .../admin/model/CanalInstanceConfig.java | 12 +- .../otter/canal/admin/model/NodeServer.java | 41 +-- .../admin/service/CanalInstanceService.java | 6 +- .../impl/CanalInstanceServiceImpl.java | 19 +- .../service/impl/NodeServerServiceImpl.java | 16 +- .../src/main/resources/application.yml | 4 + .../src/main/resources/canal_manager.sql | 18 +- .../src/main/resources/public/index.html | 2 +- .../static/js/chunk-e1a839e4.7ee86dd8.js | 1 - .../static/js/chunk-e1a839e4.e9eb1b06.js | 1 + .../src/views/canalServer/NodeServer.vue | 40 ++- client-adapter/launcher/pom.xml | 2 +- deployer/manager_ddl.sql | 37 --- .../otter/canal/deployer/CanalConstants.java | 1 + .../otter/canal/deployer/CanalController.java | 84 +++--- .../otter/canal/deployer/CanalLauncher.java | 79 +++--- .../deployer/admin/CanalAdminController.java | 22 +- .../monitor/ManagerInstanceConfigMonitor.java | 171 +++++++++++- .../monitor/SpringInstanceConfigMonitor.java | 4 - .../deployer/monitor/remote/ConfigItem.java | 47 ---- .../monitor/remote/DbRemoteConfigLoader.java | 264 ------------------ .../remote/RemoteCanalConfigMonitor.java | 13 - .../monitor/remote/RemoteConfigLoader.java | 36 --- .../remote/RemoteConfigLoaderFactory.java | 42 --- .../monitor/remote/RemoteInstanceMonitor.java | 31 -- .../remote/RemoteInstanceMonitorImpl.java | 55 ---- .../remote/http/HttpRemoteConfigLoader.java | 257 ----------------- .../monitor/remote/http/ResponseModel.java | 38 --- deployer/src/main/resources/canal.properties | 10 +- instance/manager/pom.xml | 11 + .../manager/PlainCanalInstanceGenerator.java | 72 +++++ .../instance/manager/plain}/HttpHelper.java | 67 +++-- .../instance/manager/plain/PlainCanal.java | 55 ++++ .../manager/plain/PlainCanalConfigClient.java | 141 ++++++++++ .../PlainCanalConfigClientIntegration.java | 29 ++ .../spring/SpringCanalInstanceGenerator.java | 45 ++- .../PropertyPlaceholderConfigurer.java | 45 ++- .../otter/canal/protocol/SecurityUtil.java | 21 ++ 48 files changed, 1255 insertions(+), 1025 deletions(-) create mode 100644 canal-admin/canal-admin-server/src/main/assembly/release.xml create mode 100644 canal-admin/canal-admin-server/src/main/bin/restart.sh create mode 100755 canal-admin/canal-admin-server/src/main/bin/startup.bat create mode 100644 canal-admin/canal-admin-server/src/main/bin/startup.sh create mode 100644 canal-admin/canal-admin-server/src/main/bin/stop.sh create mode 100644 canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/PollConfigController.java delete mode 100644 canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.7ee86dd8.js create mode 100644 canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.e9eb1b06.js delete mode 100644 deployer/manager_ddl.sql delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitorImpl.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpRemoteConfigLoader.java delete mode 100644 deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/ResponseModel.java create mode 100644 instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java rename {deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http => instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain}/HttpHelper.java (77%) create mode 100644 instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanal.java create mode 100644 instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java create mode 100644 instance/manager/src/test/java/com/alibaba/otter/canal/instance/manager/PlainCanalConfigClientIntegration.java diff --git a/canal-admin/canal-admin-server/pom.xml b/canal-admin/canal-admin-server/pom.xml index aa9c4c4c..9fcb1c86 100644 --- a/canal-admin/canal-admin-server/pom.xml +++ b/canal-admin/canal-admin-server/pom.xml @@ -8,9 +8,7 @@ 1.1.4-SNAPSHOT 4.0.0 - canal-admin-server - com.alibaba.otter @@ -199,6 +197,39 @@ + + org.apache.maven.plugins + maven-jar-plugin + 3.0.2 + + + application.yml + canal_manager.sql + + + + + maven-assembly-plugin + 2.2.1 + + + assemble + + single + + package + + + + false + false + + ${basedir}/src/main/assembly/release.xml + + canal-admin-${project.version} + ${project.basedir}/../../target + + diff --git a/canal-admin/canal-admin-server/src/main/assembly/release.xml b/canal-admin/canal-admin-server/src/main/assembly/release.xml new file mode 100644 index 00000000..f7b66376 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/assembly/release.xml @@ -0,0 +1,41 @@ + + dist + + tar.gz + + false + + + . + / + + README* + + + + ./src/main/bin + bin + + **/* + + 0755 + + + ./src/main/resources + /conf + + application.yml + canal_manager.sql + + + + ./target + /lib + + canal-admin-*.jar + + + + \ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/bin/restart.sh b/canal-admin/canal-admin-server/src/main/bin/restart.sh new file mode 100644 index 00000000..63749ce5 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/bin/restart.sh @@ -0,0 +1,5 @@ +#!/bin/bash + +sh stop.sh + +sh startup.sh diff --git a/canal-admin/canal-admin-server/src/main/bin/startup.bat b/canal-admin/canal-admin-server/src/main/bin/startup.bat new file mode 100755 index 00000000..840f0c03 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/bin/startup.bat @@ -0,0 +1,22 @@ +@echo off +@if not "%ECHO%" == "" echo %ECHO% +@if "%OS%" == "Windows_NT" setlocal + +set ENV_PATH=.\ +if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0% + +set conf_dir=%ENV_PATH%\..\conf + +set CLASSPATH=%conf_dir% +set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH% + +set JAVA_MEM_OPTS= -Xms128m -Xmx512m +set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8 +set ADAPTER_OPTS= -DappName=canal-admin + +set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %ADAPTER_OPTS% + +set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.admin.CanalAdminApplication +echo start cmd : %CMD_STR% + +java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.admin.CanalAdminApplication \ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/bin/startup.sh b/canal-admin/canal-admin-server/src/main/bin/startup.sh new file mode 100644 index 00000000..ec18c408 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/bin/startup.sh @@ -0,0 +1,84 @@ +#!/bin/bash + +current_path=`pwd` +case "`uname`" in + Linux) + bin_abs_path=$(readlink -f $(dirname $0)) + ;; + *) + bin_abs_path=`cd $(dirname $0); pwd` + ;; +esac +base=${bin_abs_path}/.. +export LANG=en_US.UTF-8 +export BASE=$base + +if [ -f $base/bin/adapter.pid ] ; then + echo "found adapter.pid , Please run stop.sh first ,then startup.sh" 2>&2 + exit 1 +fi + +if [ ! -d $base/logs ] ; then + mkdir -p $base/logs +fi + +## set java path +if [ -z "$JAVA" ] ; then + JAVA=$(which java) +fi + +ALIBABA_JAVA="/usr/alibaba/java/bin/java" +TAOBAO_JAVA="/opt/taobao/java/bin/java" +if [ -z "$JAVA" ]; then + if [ -f $ALIBABA_JAVA ] ; then + JAVA=$ALIBABA_JAVA + elif [ -f $TAOBAO_JAVA ] ; then + JAVA=$TAOBAO_JAVA + else + echo "Cannot find a Java JDK. Please set either set JAVA or put java (>=1.5) in your PATH." 2>&2 + exit 1 + fi +fi + +case "$#" +in +0 ) + ;; +2 ) + if [ "$1" = "debug" ]; then + DEBUG_PORT=$2 + DEBUG_SUSPEND="n" + JAVA_DEBUG_OPT="-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=$DEBUG_PORT,server=y,suspend=$DEBUG_SUSPEND" + fi + ;; +* ) + echo "THE PARAMETERS MUST BE TWO OR LESS.PLEASE CHECK AGAIN." + exit;; +esac + +str=`file -L $JAVA | grep 64-bit` +if [ -n "$str" ]; then + JAVA_OPTS="-server -Xms2048m -Xmx3072m" +else + JAVA_OPTS="-server -Xms1024m -Xmx1024m" +fi + +JAVA_OPTS="$JAVA_OPTS -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent -XX:+PrintAdaptiveSizePolicy -XX:+PrintTenuringDistribution" +JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8" +CANAL_OPTS="-DappName=canal-admin" + +for i in $base/lib/*; + do CLASSPATH=$i:"$CLASSPATH"; +done + +CLASSPATH="$base/conf:$CLASSPATH"; + +echo "cd to $bin_abs_path for workaround relative path" +cd $bin_abs_path + +echo CLASSPATH :$CLASSPATH +$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.admin.CanalAdminApplication 1>>/dev/null 2>&1 & +echo $! > $base/bin/admin.pid + +echo "cd to $current_path for continue" +cd $current_path \ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/bin/stop.sh b/canal-admin/canal-admin-server/src/main/bin/stop.sh new file mode 100644 index 00000000..05415ad9 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/bin/stop.sh @@ -0,0 +1,65 @@ +#!/bin/bash + +cygwin=false; +linux=false; +case "`uname`" in + CYGWIN*) + cygwin=true + ;; + Linux*) + linux=true + ;; +esac + +get_pid() { + STR=$1 + PID=$2 + if $cygwin; then + JAVA_CMD="$JAVA_HOME\bin\java" + JAVA_CMD=`cygpath --path --unix $JAVA_CMD` + JAVA_PID=`ps |grep $JAVA_CMD |awk '{print $1}'` + else + if $linux; then + if [ ! -z "$PID" ]; then + JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` + else + JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'` + fi + else + if [ ! -z "$PID" ]; then + JAVA_PID=`ps aux |grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` + else + JAVA_PID=`ps aux |grep "$STR"|grep -v grep|awk '{print $2}'` + fi + fi + fi + echo $JAVA_PID; +} + +base=`dirname $0`/.. +pidfile=$base/bin/admin.pid +if [ ! -f "$pidfile" ];then + echo "canal-admin is not running. exists" + exit +fi + +pid=`cat $pidfile` +if [ "$pid" == "" ] ; then + pid=`get_pid "appName=canal-admin"` +fi + +echo -e "`hostname`: stopping canal $pid ... " +kill $pid + +LOOPS=0 +while (true); +do + gpid=`get_pid "appName=canal-admin" "$pid"` + if [ "$gpid" == "" ] ; then + echo "Oook! cost:$LOOPS" + `rm $pidfile` + break; + fi + let LOOPS=LOOPS+1 + sleep 1 +done \ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java index fde7d6b4..07512fce 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/config/WebConfig.java @@ -74,8 +74,8 @@ public class WebConfig implements WebMvcConfigurer { httpServletResponse.setContentType("application/json;charset=UTF-8"); PrintWriter out = httpServletResponse.getWriter(); out.print(json); - } catch (Exception e) { - e.printStackTrace(); + } catch (Throwable e) { + throw new RuntimeException(e); } return false; } @@ -84,6 +84,7 @@ public class WebConfig implements WebMvcConfigurer { } }) .addPathPatterns("/api/**") + .excludePathPatterns("/api/**/config/**") .excludePathPatterns("/api/**/user/login") .excludePathPatterns("/api/**/user/logout") .excludePathPatterns("/api/**/user/info"); diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java index 1bc65b9a..3ebc392a 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/CanalConfigController.java @@ -1,7 +1,12 @@ package com.alibaba.otter.canal.admin.controller; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.web.bind.annotation.*; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PutMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; import com.alibaba.otter.canal.admin.model.BaseModel; import com.alibaba.otter.canal.admin.model.CanalConfig; @@ -31,17 +36,6 @@ public class CanalConfigController { return BaseModel.getInstance(canalConfigService.getCanalConfig()); } - /** - * 获取配置信息摘要(无配置内容) - * - * @param env 环境变量 - * @return 配置信息摘要 - */ - @GetMapping(value = "/config/summary") - public BaseModel canalConfigSummary(@PathVariable String env) { - return BaseModel.getInstance(canalConfigService.getCanalConfigSummary()); - } - /** * 修改配置 * @@ -54,4 +48,5 @@ public class CanalConfigController { canalConfigService.updateContent(canalConfig); return BaseModel.getInstance("success"); } + } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/PollConfigController.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/PollConfigController.java new file mode 100644 index 00000000..0ac74ef1 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/controller/PollConfigController.java @@ -0,0 +1,155 @@ +package com.alibaba.otter.canal.admin.controller; + +import java.security.NoSuchAlgorithmException; +import java.util.List; +import java.util.stream.Collectors; + +import org.apache.commons.lang.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import com.alibaba.otter.canal.admin.model.BaseModel; +import com.alibaba.otter.canal.admin.model.CanalConfig; +import com.alibaba.otter.canal.admin.model.CanalInstanceConfig; +import com.alibaba.otter.canal.admin.service.CanalConfigService; +import com.alibaba.otter.canal.admin.service.CanalInstanceService; +import com.alibaba.otter.canal.protocol.SecurityUtil; +import com.google.common.base.Joiner; + +/** + * Canal Instance配置管理控制层 + * + * @author rewerma 2019-07-13 下午05:12:16 + * @version 1.0.0 + */ +@RestController +@RequestMapping("/api/{env}/config") +public class PollConfigController { + + private static final byte[] seeds = "canal is best!".getBytes(); + + @Autowired + CanalInstanceService canalInstanceConfigService; + + @Autowired + CanalConfigService canalConfigService; + + @Value(value = "${canal.adminUser}") + String user; + + @Value(value = "${canal.adminPasswd}") + String passwd; + + /** + * 获取server全局配置 + */ + @GetMapping(value = "/server_poll") + public BaseModel canalConfigPoll(@RequestHeader String user, @RequestHeader String passwd, + @PathVariable String env, @RequestParam String md5) { + if (!auth(user, passwd)) { + throw new RuntimeException("auth :" + user + " is failed"); + } + + CanalConfig config = canalConfigService.getCanalConfig(); + if (StringUtils.isEmpty(md5)) { + return BaseModel.getInstance(config); + } else { + + try { + String newMd5 = SecurityUtil.md5String(config.getContent()); + if (StringUtils.equals(md5, newMd5)) { + config.setContent(null); + } + } catch (NoSuchAlgorithmException e) { + } + + return BaseModel.getInstance(config); + } + } + + /** + * 获取单个instance的配置 + */ + @GetMapping(value = "/instance_poll/{destination}") + public BaseModel instanceConfigPoll(@RequestHeader String user, @RequestHeader String passwd, + @PathVariable String env, + @PathVariable String destination, @RequestParam String md5) { + if (!auth(user, passwd)) { + throw new RuntimeException("auth :" + user + " is failed"); + } + + CanalInstanceConfig config = canalInstanceConfigService.findOne(destination); + if (StringUtils.isEmpty(md5)) { + return BaseModel.getInstance(config); + } else { + try { + String newMd5 = SecurityUtil.md5String(config.getContent()); + if (StringUtils.equals(md5, newMd5)) { + config.setContent(null); + } + } catch (NoSuchAlgorithmException e) { + } + + return BaseModel.getInstance(config); + } + } + + /** + * 获取对应server(ip+port)所需要运行的instance列表 + */ + @GetMapping(value = "/instances_poll") + public BaseModel instancesPoll(@RequestHeader String user, @RequestHeader String passwd, + @PathVariable String env, @RequestParam String ip, + @RequestParam String port, @RequestParam String md5) { + if (!auth(user, passwd)) { + throw new RuntimeException("auth :" + user + " is failed"); + } + + CanalInstanceConfig canalInstanceConfig = new CanalInstanceConfig(); + List configs = canalInstanceConfigService.findList(canalInstanceConfig); + List instances = configs.stream().map(config -> config.getName()).collect(Collectors.toList()); + String data = Joiner.on(',').join(instances); + canalInstanceConfig.setContent(data); + if (StringUtils.isEmpty(md5)) { + return BaseModel.getInstance(canalInstanceConfig); + } else { + try { + String newMd5 = SecurityUtil.md5String(canalInstanceConfig.getContent()); + if (StringUtils.equals(md5, newMd5)) { + canalInstanceConfig.setContent(null); + } + } catch (NoSuchAlgorithmException e) { + } + + return BaseModel.getInstance(canalInstanceConfig); + } + } + + private boolean auth(String user, String passwd) { + // 如果user/passwd密码为空,则任何用户账户都能登录 + if ((StringUtils.isEmpty(this.user) || StringUtils.equals(this.user, user))) { + if (StringUtils.isEmpty(this.passwd)) { + return true; + } else if (StringUtils.isEmpty(passwd)) { + // 如果server密码有配置,客户端密码为空,则拒绝 + return false; + } + + try { + // manager这里保存了原始密码，反过来和canal发送过来的进行校验 + byte[] passForClient = SecurityUtil.scramble411(this.passwd.getBytes(), seeds); + return SecurityUtil.scrambleServerAuth(passForClient, SecurityUtil.hexStr2Bytes(passwd), seeds); + } catch (NoSuchAlgorithmException e) { + return false; + } + } + + return false; + } +} diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java index f9220afe..409d07e7 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalConfig.java @@ -33,6 +33,7 @@ public class CanalConfig extends Model { private Long id; private String name; private String content; + private String status; private Date modifiedTime; public Long getId() { @@ -66,4 +67,13 @@ public class CanalConfig extends Model { public void setModifiedTime(Date modifiedTime) { this.modifiedTime = modifiedTime; } + + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalInstanceConfig.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalInstanceConfig.java index 490c0ea6..e103a1e6 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalInstanceConfig.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/CanalInstanceConfig.java @@ -2,10 +2,11 @@ package com.alibaba.otter.canal.admin.model; import io.ebean.Finder; +import java.util.Date; + import javax.persistence.Entity; import javax.persistence.Id; import javax.persistence.Transient; -import java.util.Date; /** * Canal实例配置信息实体类 @@ -33,6 +34,7 @@ public class CanalInstanceConfig extends Model { private Long id; private String name; private String content; + private String status; private Date modifiedTime; @Transient @@ -64,6 +66,14 @@ public class CanalInstanceConfig extends Model { this.content = content; } + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + public Date getModifiedTime() { return modifiedTime; } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java index 20369c05..1f9834ee 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/model/NodeServer.java @@ -1,13 +1,13 @@ package com.alibaba.otter.canal.admin.model; +import io.ebean.Finder; + import java.util.Date; import javax.persistence.Entity; import javax.persistence.Id; import javax.persistence.Table; -import io.ebean.Finder; - /** * 节点信息实体类 * @@ -35,13 +35,14 @@ public class NodeServer extends Model { private Long id; private String name; private String ip; - private Integer port; - private Integer port2; - private Integer status; + private Integer adminPort; + private Integer metricPort; + private Integer tcpPort; + private String status; private Date modifiedTime; public void init() { - status = -1; + status = "-1"; } public Long getId() { @@ -68,27 +69,35 @@ public class NodeServer extends Model { this.ip = ip; } - public Integer getPort() { - return port; + public Integer getAdminPort() { + return adminPort; } - public void setPort(Integer port) { - this.port = port; + public void setAdminPort(Integer adminPort) { + this.adminPort = adminPort; } - public Integer getPort2() { - return port2; + public Integer getMetricPort() { + return metricPort; } - public void setPort2(Integer port2) { - this.port2 = port2; + public void setMetricPort(Integer metricPort) { + this.metricPort = metricPort; } - public Integer getStatus() { + public Integer getTcpPort() { + return tcpPort; + } + + public void setTcpPort(Integer tcpPort) { + this.tcpPort = tcpPort; + } + + public String getStatus() { return status; } - public void setStatus(Integer status) { + public void setStatus(String status) { this.status = status; } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalInstanceService.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalInstanceService.java index 5f7ae443..1bc9c29a 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalInstanceService.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/CanalInstanceService.java @@ -1,10 +1,10 @@ package com.alibaba.otter.canal.admin.service; -import com.alibaba.otter.canal.admin.model.CanalInstanceConfig; - import java.util.List; import java.util.Map; +import com.alibaba.otter.canal.admin.model.CanalInstanceConfig; + /** * Canal实例配置信息业务层接口 * @@ -19,6 +19,8 @@ public interface CanalInstanceService { CanalInstanceConfig detail(Long id); + CanalInstanceConfig findOne(String name); + void updateContent(CanalInstanceConfig canalInstanceConfig); void delete(Long id); diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalInstanceServiceImpl.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalInstanceServiceImpl.java index 659eff08..f60d767e 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalInstanceServiceImpl.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/CanalInstanceServiceImpl.java @@ -40,7 +40,7 @@ public class CanalInstanceServiceImpl implements CanalInstanceService { List nodeServers = NodeServer.find.query().findList(); for (NodeServer nodeServer : nodeServers) { String runningInstances = SimpleAdminConnectors.execute(nodeServer.getIp(), - nodeServer.getPort(), + nodeServer.getAdminPort(), AdminConnector::getRunningInstances); if (runningInstances == null) { continue; @@ -79,6 +79,16 @@ public class CanalInstanceServiceImpl implements CanalInstanceService { } } + @Override + public CanalInstanceConfig findOne(String name) { + CanalInstanceConfig config = CanalInstanceConfig.find.query() + .setDisableLazyLoading(true) + .where() + .eq("name", name) + .findOne(); + return config; + } + public Map remoteInstanceLog(Long id, Long nodeId) { Map result = new HashMap<>(); @@ -92,7 +102,7 @@ public class CanalInstanceServiceImpl implements CanalInstanceService { } String log = SimpleAdminConnectors.execute(nodeServer.getIp(), - nodeServer.getPort(), + nodeServer.getAdminPort(), adminConnector -> adminConnector.instanceLog(canalInstanceConfig.getName(), null, 100)); result.put("instance", canalInstanceConfig.getName()); @@ -124,11 +134,11 @@ public class CanalInstanceServiceImpl implements CanalInstanceService { Boolean resutl = null; if ("start".equals(option)) { resutl = SimpleAdminConnectors.execute(nodeServer.getIp(), - nodeServer.getPort(), + nodeServer.getAdminPort(), adminConnector -> adminConnector.startInstance(canalInstanceConfig.getName())); } else if ("stop".equals(option)) { resutl = SimpleAdminConnectors.execute(nodeServer.getIp(), - nodeServer.getPort(), + nodeServer.getAdminPort(), adminConnector -> adminConnector.stopInstance(canalInstanceConfig.getName())); } else { return false; @@ -139,4 +149,5 @@ public class CanalInstanceServiceImpl implements CanalInstanceService { } return resutl; } + } diff --git a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java index d6b2ce8b..5f9610af 100644 --- a/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java +++ b/canal-admin/canal-admin-server/src/main/java/com/alibaba/otter/canal/admin/service/impl/NodeServerServiceImpl.java @@ -31,7 +31,7 @@ public class NodeServerServiceImpl implements NodeServerService { int cnt = NodeServer.find.query() .where() .eq("ip", nodeServer.getIp()) - .eq("port", nodeServer.getPort()) + .eq("admin_port", nodeServer.getAdminPort()) .findCount(); if (cnt > 0) { throw new ServiceException("节点信息已存在"); @@ -48,14 +48,14 @@ public class NodeServerServiceImpl implements NodeServerService { int cnt = NodeServer.find.query() .where() .eq("ip", nodeServer.getIp()) - .eq("port", nodeServer.getPort()) + .eq("admin_port", nodeServer.getAdminPort()) .ne("id", nodeServer.getId()) .findCount(); if (cnt > 0) { throw new ServiceException("节点信息已存在"); } - nodeServer.update("name", "ip", "port", "port2"); + nodeServer.update("name", "ip", "admin_port", "tcp_port", "metric_port"); } public void delete(Long id) { @@ -86,8 +86,8 @@ public class NodeServerServiceImpl implements NodeServerService { // get all nodes status for (NodeServer ns : nodeServers) { futures.add(executorService.submit(() -> { - boolean status = SimpleAdminConnectors.execute(ns.getIp(), ns.getPort(), AdminConnector::check); - ns.setStatus(status ? 1 : 0); + boolean status = SimpleAdminConnectors.execute(ns.getIp(), ns.getAdminPort(), AdminConnector::check); + ns.setStatus(status ? "1" : "0"); return !status; })); } @@ -115,7 +115,7 @@ public class NodeServerServiceImpl implements NodeServerService { return ""; } return SimpleAdminConnectors.execute(nodeServer.getIp(), - nodeServer.getPort(), + nodeServer.getAdminPort(), adminConnector -> adminConnector.canalLog(100)); } @@ -126,9 +126,9 @@ public class NodeServerServiceImpl implements NodeServerService { } Boolean result = null; if ("start".equals(option)) { - result = SimpleAdminConnectors.execute(nodeServer.getIp(), nodeServer.getPort(), AdminConnector::start); + result = SimpleAdminConnectors.execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::start); } else if ("stop".equals(option)) { - result = SimpleAdminConnectors.execute(nodeServer.getIp(), nodeServer.getPort(), AdminConnector::stop); + result = SimpleAdminConnectors.execute(nodeServer.getIp(), nodeServer.getAdminPort(), AdminConnector::stop); } else { return false; } diff --git a/canal-admin/canal-admin-server/src/main/resources/application.yml b/canal-admin/canal-admin-server/src/main/resources/application.yml index d241e35a..7adc1c7a 100644 --- a/canal-admin/canal-admin-server/src/main/resources/application.yml +++ b/canal-admin/canal-admin-server/src/main/resources/application.yml @@ -13,3 +13,7 @@ spring.datasource: hikari: maximum-pool-size: 10 minimum-idle: 1 + +canal: + adminUser: admin + adminPasswd: admin diff --git a/canal-admin/canal-admin-server/src/main/resources/canal_manager.sql b/canal-admin/canal-admin-server/src/main/resources/canal_manager.sql index 9ca6de4f..a926380d 100644 --- a/canal-admin/canal-admin-server/src/main/resources/canal_manager.sql +++ b/canal-admin/canal-admin-server/src/main/resources/canal_manager.sql @@ -13,10 +13,11 @@ CREATE TABLE `canal_adapter_config` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `category` varchar(45) NOT NULL, `name` varchar(45) NOT NULL, + `status` varchar(45) DEFAULT NULL, `content` text NOT NULL, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`) -) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4; -- ---------------------------- -- Table structure for canal_config @@ -25,11 +26,12 @@ DROP TABLE IF EXISTS `canal_config`; CREATE TABLE `canal_config` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `name` varchar(45) NOT NULL, + `status` varchar(45) DEFAULT NULL, `content` text NOT NULL, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `name_UNIQUE` (`name`) -) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4; -- ---------------------------- -- Table structure for canal_instance_config @@ -38,11 +40,12 @@ DROP TABLE IF EXISTS `canal_instance_config`; CREATE TABLE `canal_instance_config` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `name` varchar(45) NOT NULL, + `status` varchar(45) DEFAULT NULL, `content` text NOT NULL, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `name_UNIQUE` (`name`) -) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4; -- ---------------------------- -- Table structure for canal_node_server @@ -52,9 +55,10 @@ CREATE TABLE `canal_node_server` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `name` varchar(63) NOT NULL, `ip` varchar(63) NOT NULL, - `port` int(11) DEFAULT NULL, - `port2` int(11) DEFAULT NULL, - `status` int(11) NOT NULL, + `admin_port` int(11) DEFAULT NULL, + `tcp_port` int(11) DEFAULT NULL, + `metric_port` int(11) DEFAULT NULL, + `status` varchar(45) DEFAULT NULL, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4; @@ -73,7 +77,7 @@ CREATE TABLE `canal_user` ( `avatar` varchar(255) DEFAULT NULL, `creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`) -) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4; -- ---------------------------- -- Records of canal_user diff --git a/canal-admin/canal-admin-server/src/main/resources/public/index.html b/canal-admin/canal-admin-server/src/main/resources/public/index.html index b76917f2..6f6beecc 100644 --- a/canal-admin/canal-admin-server/src/main/resources/public/index.html +++ b/canal-admin/canal-admin-server/src/main/resources/public/index.html @@ -1 +1 @@ -Canal Admin
\ No newline at end of file +Canal Admin
\ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.7ee86dd8.js b/canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.7ee86dd8.js deleted file mode 100644 index 33f85501..00000000 --- a/canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.7ee86dd8.js +++ /dev/null @@ -1 +0,0 @@ -(window["webpackJsonp"]=window["webpackJsonp"]||[]).push([["chunk-e1a839e4"],{"9f66":function(e,t,n){"use strict";n.r(t);var r=function(){var e=this,t=e.$createElement,n=e._self._c||t;return n("div",{staticClass:"app-container"},[n("div",{staticClass:"filter-container"},[n("el-button",{staticClass:"filter-item",attrs:{type:"primary"},on:{click:function(t){return e.handleCreate()}}},[e._v("新建Server")]),e._v(" "),n("el-button",{staticClass:"filter-item",attrs:{type:"info"},on:{click:function(t){return e.fetchData()}}},[e._v("刷新列表")])],1),e._v(" "),n("el-table",{directives:[{name:"loading",rawName:"v-loading",value:e.listLoading,expression:"listLoading"}],attrs:{data:e.list,"element-loading-text":"Loading",border:"",fit:"","highlight-current-row":""}},[n("el-table-column",{attrs:{label:"Server 名称","min-width":"200",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.name)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"Server IP","min-width":"200",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[n("span",[e._v(e._s(t.row.ip))])]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"Admin 端口","min-width":"100",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.port)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"监控端口","min-width":"100",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.port2)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{"class-name":"status-col",label:"状态","min-width":"150",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[n("el-tag",{attrs:{type:e._f("statusFilter")(t.row.status)}},[e._v(e._s(e._f("statusLabel")(t.row.status)))])]}}])}),e._v(" "),n("el-table-column",{attrs:{align:"center",prop:"created_at",label:"操作","min-width":"150"},scopedSlots:e._u([{key:"default",fn:function(t){return[n("el-dropdown",{attrs:{trigger:"click"}},[n("el-button",{attrs:{type:"primary",size:"mini"}},[e._v("\n 操作"),n("i",{staticClass:"el-icon-arrow-down el-icon--right"})]),e._v(" "),n("el-dropdown-menu",{attrs:{slot:"dropdown"},slot:"dropdown"},[n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleUpdate(t.row)}}},[e._v("修改")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleDelete(t.row)}}},[e._v("删除")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleStart(t.row)}}},[e._v("启动")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleStop(t.row)}}},[e._v("停止")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleLog(t.row)}}},[e._v("日志")])],1)],1)]}}])})],1),e._v(" "),n("el-dialog",{attrs:{visible:e.dialogFormVisible,title:e.textMap[e.dialogStatus],width:"600px"},on:{"update:visible":function(t){e.dialogFormVisible=t}}},[n("el-form",{ref:"dataForm",staticStyle:{width:"400px","margin-left":"30px"},attrs:{rules:e.rules,model:e.nodeModel,"label-position":"left","label-width":"120px"}},[n("el-form-item",{attrs:{label:"Server 名称",prop:"name"}},[n("el-input",{model:{value:e.nodeModel.name,callback:function(t){e.$set(e.nodeModel,"name",t)},expression:"nodeModel.name"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"Server IP",prop:"ip"}},[n("el-input",{model:{value:e.nodeModel.ip,callback:function(t){e.$set(e.nodeModel,"ip",t)},expression:"nodeModel.ip"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"Admin 端口",prop:"port"}},[n("el-input",{attrs:{placeholder:"11110",type:"number"},model:{value:e.nodeModel.port,callback:function(t){e.$set(e.nodeModel,"port",t)},expression:"nodeModel.port"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"监控端口",prop:"port2"}},[n("el-input",{attrs:{placeholder:"11112",type:"number"},model:{value:e.nodeModel.port2,callback:function(t){e.$set(e.nodeModel,"port2",t)},expression:"nodeModel.port2"}})],1)],1),e._v(" "),n("div",{staticClass:"dialog-footer",attrs:{slot:"footer"},slot:"footer"},[n("el-button",{on:{click:function(t){e.dialogFormVisible=!1}}},[e._v("取消")]),e._v(" "),n("el-button",{attrs:{type:"primary"},on:{click:function(t){return e.dataOperation()}}},[e._v("确定")])],1)],1)],1)},o=[],a=n("c6ed"),i={filters:{statusFilter:function(e){var t={1:"success",0:"gray","-1":"danger"};return t[e]},statusLabel:function(e){var t={1:"启动",0:"停止","-1":"断开"};return t[e]}},data:function(){return{list:null,listLoading:!0,listQuery:{name:"",ip:""},dialogFormVisible:!1,textMap:{create:"新建Server信息",update:"修改Server信息"},nodeModel:{id:void 0,name:null,ip:null,port:11110,port2:11112},rules:{name:[{required:!0,message:"Server 名称不能为空",trigger:"change"}],ip:[{required:!0,message:"Server IP不能为空",trigger:"change"}],port:[{required:!0,message:"Server Admin端口不能为空",trigger:"change"}]},dialogStatus:"create"}},created:function(){this.fetchData()},methods:{fetchData:function(){var e=this;this.listLoading=!0,Object(a["c"])(this.listQuery).then(function(t){e.list=t.data,e.listLoading=!1})},resetModel:function(){this.nodeModel={id:void 0,name:null,ip:null,port:null,port2:null}},handleCreate:function(){var e=this;this.resetModel(),this.dialogStatus="create",this.dialogFormVisible=!0,this.$nextTick(function(){e.$refs["dataForm"].clearValidate()})},dataOperation:function(){var e=this;this.$refs["dataForm"].validate(function(t){t&&("create"===e.dialogStatus&&Object(a["a"])(e.nodeModel).then(function(t){e.operationRes(t)}),"update"===e.dialogStatus&&Object(a["g"])(e.nodeModel).then(function(t){e.operationRes(t)}))})},operationRes:function(e){"success"===e.data?(this.fetchData(),this.dialogFormVisible=!1,this.$message({message:this.textMap[this.dialogStatus]+"成功",type:"success"})):this.$message({message:this.textMap[this.dialogStatus]+"失败",type:"error"})},handleUpdate:function(e){var t=this;this.resetModel(),this.nodeModel=Object.assign({},e),this.dialogStatus="update",this.dialogFormVisible=!0,this.$nextTick(function(){t.$refs["dataForm"].clearValidate()})},handleDelete:function(e){var t=this;this.$confirm("删除Server信息并不会导致节点服务停止","确定删除Server信息",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){Object(a["b"])(e.id).then(function(e){"success"===e.data?(t.fetchData(),t.$message({message:"删除Server信息成功",type:"success"})):t.$message({message:"删除Server信息失败",type:"error"})})})},handleStart:function(e){var t=this;0===e.status?this.$confirm("启动Server服务","确定启动Server服务",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){Object(a["e"])(e.id).then(function(e){e.data?(t.fetchData(),t.$message({message:"启动成功",type:"success"})):t.$message({message:"启动Server服务出现异常",type:"error"})})}):this.$message({message:"当前Server不是停止状态，无法启动",type:"error"})},handleStop:function(e){var t=this;1===e.status?this.$confirm("停止 Server 服务","确定停止Server服务",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){Object(a["f"])(e.id).then(function(e){e.data?(t.fetchData(),t.$message({message:"停止成功",type:"success"})):t.$message({message:"停止Server服务出现异常",type:"error"})})}):this.$message({message:"当前Server不是启动状态，无法停止",type:"error"})},handleLog:function(e){this.$router.push("nodeServer/log?id="+e.id)}}},l=i,s=n("2877"),d=Object(s["a"])(l,r,o,!1,null,null,null);t["default"]=d.exports},c6ed:function(e,t,n){"use strict";n.d(t,"c",function(){return o}),n.d(t,"a",function(){return a}),n.d(t,"g",function(){return i}),n.d(t,"b",function(){return l}),n.d(t,"e",function(){return s}),n.d(t,"f",function(){return d}),n.d(t,"d",function(){return u});var r=n("b775");function o(e){return Object(r["a"])({url:"/nodeServers",method:"get",params:e})}function a(e){return Object(r["a"])({url:"/nodeServer",method:"post",data:e})}function i(e){return Object(r["a"])({url:"/nodeServer",method:"put",data:e})}function l(e){return Object(r["a"])({url:"/nodeServer/"+e,method:"delete"})}function s(e){return Object(r["a"])({url:"/nodeServer/start/"+e,method:"put"})}function d(e){return Object(r["a"])({url:"/nodeServer/stop/"+e,method:"put"})}function u(e){return Object(r["a"])({url:"/nodeServer/log/"+e,method:"get"})}}}]); \ No newline at end of file diff --git a/canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.e9eb1b06.js b/canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.e9eb1b06.js new file mode 100644 index 00000000..33ec1b50 --- /dev/null +++ b/canal-admin/canal-admin-server/src/main/resources/public/static/js/chunk-e1a839e4.e9eb1b06.js @@ -0,0 +1 @@ +(window["webpackJsonp"]=window["webpackJsonp"]||[]).push([["chunk-e1a839e4"],{"9f66":function(e,t,n){"use strict";n.r(t);var r=function(){var e=this,t=e.$createElement,n=e._self._c||t;return n("div",{staticClass:"app-container"},[n("div",{staticClass:"filter-container"},[n("el-button",{staticClass:"filter-item",attrs:{type:"primary"},on:{click:function(t){return e.handleCreate()}}},[e._v("新建Server")]),e._v(" "),n("el-button",{staticClass:"filter-item",attrs:{type:"info"},on:{click:function(t){return e.fetchData()}}},[e._v("刷新列表")])],1),e._v(" "),n("el-table",{directives:[{name:"loading",rawName:"v-loading",value:e.listLoading,expression:"listLoading"}],attrs:{data:e.list,"element-loading-text":"Loading",border:"",fit:"","highlight-current-row":""}},[n("el-table-column",{attrs:{label:"Server 名称","min-width":"200",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.name)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"Server IP","min-width":"200",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[n("span",[e._v(e._s(t.row.ip))])]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"admin 端口","min-width":"100",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.adminPort)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"tcp 端口","min-width":"100",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.tcpPort)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{label:"metric 端口","min-width":"100",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[e._v("\n "+e._s(t.row.metricPort)+"\n ")]}}])}),e._v(" "),n("el-table-column",{attrs:{"class-name":"status-col",label:"状态","min-width":"150",align:"center"},scopedSlots:e._u([{key:"default",fn:function(t){return[n("el-tag",{attrs:{type:e._f("statusFilter")(t.row.status)}},[e._v(e._s(e._f("statusLabel")(t.row.status)))])]}}])}),e._v(" "),n("el-table-column",{attrs:{align:"center",prop:"created_at",label:"操作","min-width":"150"},scopedSlots:e._u([{key:"default",fn:function(t){return[n("el-dropdown",{attrs:{trigger:"click"}},[n("el-button",{attrs:{type:"primary",size:"mini"}},[e._v("\n 操作"),n("i",{staticClass:"el-icon-arrow-down el-icon--right"})]),e._v(" "),n("el-dropdown-menu",{attrs:{slot:"dropdown"},slot:"dropdown"},[n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleUpdate(t.row)}}},[e._v("修改")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleDelete(t.row)}}},[e._v("删除")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleStart(t.row)}}},[e._v("启动")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleStop(t.row)}}},[e._v("停止")]),e._v(" "),n("el-dropdown-item",{nativeOn:{click:function(n){return e.handleLog(t.row)}}},[e._v("日志")])],1)],1)]}}])})],1),e._v(" "),n("el-dialog",{attrs:{visible:e.dialogFormVisible,title:e.textMap[e.dialogStatus],width:"600px"},on:{"update:visible":function(t){e.dialogFormVisible=t}}},[n("el-form",{ref:"dataForm",staticStyle:{width:"400px","margin-left":"30px"},attrs:{rules:e.rules,model:e.nodeModel,"label-position":"left","label-width":"120px"}},[n("el-form-item",{attrs:{label:"Server 名称",prop:"name"}},[n("el-input",{model:{value:e.nodeModel.name,callback:function(t){e.$set(e.nodeModel,"name",t)},expression:"nodeModel.name"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"Server IP",prop:"ip"}},[n("el-input",{model:{value:e.nodeModel.ip,callback:function(t){e.$set(e.nodeModel,"ip",t)},expression:"nodeModel.ip"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"admin 端口",prop:"adminPort"}},[n("el-input",{attrs:{placeholder:"11110",type:"number"},model:{value:e.nodeModel.adminPort,callback:function(t){e.$set(e.nodeModel,"adminPort",t)},expression:"nodeModel.adminPort"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"tcp 端口",prop:"tcpPort"}},[n("el-input",{attrs:{placeholder:"11111",type:"number"},model:{value:e.nodeModel.tcpPort,callback:function(t){e.$set(e.nodeModel,"tcpPort",t)},expression:"nodeModel.tcpPort"}})],1),e._v(" "),n("el-form-item",{attrs:{label:"metric 端口",prop:"metricPort"}},[n("el-input",{attrs:{placeholder:"11112",type:"number"},model:{value:e.nodeModel.metricPort,callback:function(t){e.$set(e.nodeModel,"metricPort",t)},expression:"nodeModel.metricPort"}})],1)],1),e._v(" "),n("div",{staticClass:"dialog-footer",attrs:{slot:"footer"},slot:"footer"},[n("el-button",{on:{click:function(t){e.dialogFormVisible=!1}}},[e._v("取消")]),e._v(" "),n("el-button",{attrs:{type:"primary"},on:{click:function(t){return e.dataOperation()}}},[e._v("确定")])],1)],1)],1)},o=[],a=n("c6ed"),i={filters:{statusFilter:function(e){var t={1:"success",0:"gray","-1":"danger"};return t[e]},statusLabel:function(e){var t={1:"启动",0:"停止","-1":"断开"};return t[e]}},data:function(){return{list:null,listLoading:!0,listQuery:{name:"",ip:""},dialogFormVisible:!1,textMap:{create:"新建Server信息",update:"修改Server信息"},nodeModel:{id:void 0,name:null,ip:null,adminPort:11110,tcpPort:11111,metricPort:11112},rules:{name:[{required:!0,message:"Server 名称不能为空",trigger:"change"}],ip:[{required:!0,message:"Server IP不能为空",trigger:"change"}],port:[{required:!0,message:"Server admin端口不能为空",trigger:"change"}]},dialogStatus:"create"}},created:function(){this.fetchData()},methods:{fetchData:function(){var e=this;this.listLoading=!0,Object(a["c"])(this.listQuery).then(function(t){e.list=t.data,e.listLoading=!1})},resetModel:function(){this.nodeModel={id:void 0,name:null,ip:null,adminPort:null,tcpPort:null,metricPort:null}},handleCreate:function(){var e=this;this.resetModel(),this.dialogStatus="create",this.dialogFormVisible=!0,this.$nextTick(function(){e.$refs["dataForm"].clearValidate()})},dataOperation:function(){var e=this;this.$refs["dataForm"].validate(function(t){t&&("create"===e.dialogStatus&&Object(a["a"])(e.nodeModel).then(function(t){e.operationRes(t)}),"update"===e.dialogStatus&&Object(a["g"])(e.nodeModel).then(function(t){e.operationRes(t)}))})},operationRes:function(e){"success"===e.data?(this.fetchData(),this.dialogFormVisible=!1,this.$message({message:this.textMap[this.dialogStatus]+"成功",type:"success"})):this.$message({message:this.textMap[this.dialogStatus]+"失败",type:"error"})},handleUpdate:function(e){var t=this;this.resetModel(),this.nodeModel=Object.assign({},e),this.dialogStatus="update",this.dialogFormVisible=!0,this.$nextTick(function(){t.$refs["dataForm"].clearValidate()})},handleDelete:function(e){var t=this;this.$confirm("删除Server信息并不会导致节点服务停止","确定删除Server信息",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){Object(a["b"])(e.id).then(function(e){"success"===e.data?(t.fetchData(),t.$message({message:"删除Server信息成功",type:"success"})):t.$message({message:"删除Server信息失败",type:"error"})})})},handleStart:function(e){var t=this;"0"===e.status?this.$confirm("启动Server服务","确定启动Server服务",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){Object(a["e"])(e.id).then(function(e){e.data?(t.fetchData(),t.$message({message:"启动成功",type:"success"})):t.$message({message:"启动Server服务出现异常",type:"error"})})}):this.$message({message:"当前Server不是停止状态，无法启动",type:"error"})},handleStop:function(e){var t=this;"1"===e.status?this.$confirm("停止 Server 服务","确定停止Server服务",{confirmButtonText:"确定",cancelButtonText:"取消",type:"warning"}).then(function(){Object(a["f"])(e.id).then(function(e){e.data?(t.fetchData(),t.$message({message:"停止成功",type:"success"})):t.$message({message:"停止Server服务出现异常",type:"error"})})}):this.$message({message:"当前Server不是启动状态，无法停止",type:"error"})},handleLog:function(e){this.$router.push("nodeServer/log?id="+e.id)}}},l=i,s=n("2877"),c=Object(s["a"])(l,r,o,!1,null,null,null);t["default"]=c.exports},c6ed:function(e,t,n){"use strict";n.d(t,"c",function(){return o}),n.d(t,"a",function(){return a}),n.d(t,"g",function(){return i}),n.d(t,"b",function(){return l}),n.d(t,"e",function(){return s}),n.d(t,"f",function(){return c}),n.d(t,"d",function(){return d});var r=n("b775");function o(e){return Object(r["a"])({url:"/nodeServers",method:"get",params:e})}function a(e){return Object(r["a"])({url:"/nodeServer",method:"post",data:e})}function i(e){return Object(r["a"])({url:"/nodeServer",method:"put",data:e})}function l(e){return Object(r["a"])({url:"/nodeServer/"+e,method:"delete"})}function s(e){return Object(r["a"])({url:"/nodeServer/start/"+e,method:"put"})}function c(e){return Object(r["a"])({url:"/nodeServer/stop/"+e,method:"put"})}function d(e){return Object(r["a"])({url:"/nodeServer/log/"+e,method:"get"})}}}]); \ No newline at end of file diff --git a/canal-admin/canal-admin-ui/src/views/canalServer/NodeServer.vue b/canal-admin/canal-admin-ui/src/views/canalServer/NodeServer.vue index c639547f..cdc05369 100644 --- a/canal-admin/canal-admin-ui/src/views/canalServer/NodeServer.vue +++ b/canal-admin/canal-admin-ui/src/views/canalServer/NodeServer.vue @@ -25,14 +25,19 @@ {{ scope.row.ip }} - + - + + + + @@ -65,11 +70,14 @@ - - + + - - + + + + +
@@ -119,13 +127,14 @@ export default { id: undefined, name: null, ip: null, - port: 11110, - port2: 11112 + adminPort: 11110, + tcpPort: 11111, + metricPort: 11112 }, rules: { name: [{ required: true, message: 'Server 名称不能为空', trigger: 'change' }], ip: [{ required: true, message: 'Server IP不能为空', trigger: 'change' }], - port: [{ required: true, message: 'Server Admin端口不能为空', trigger: 'change' }] + port: [{ required: true, message: 'Server admin端口不能为空', trigger: 'change' }] }, dialogStatus: 'create' } @@ -147,8 +156,9 @@ export default { id: undefined, name: null, ip: null, - port: null, - port2: null + adminPort: null, + tcpPort: null, + metricPort: null } }, handleCreate() { @@ -222,7 +232,7 @@ export default { }) }, handleStart(row) { - if (row.status !== 0) { + if (row.status !== '0') { this.$message({ message: '当前Server不是停止状态，无法启动', type: 'error' }) return } @@ -248,7 +258,7 @@ export default { }) }, handleStop(row) { - if (row.status !== 1) { + if (row.status !== '1') { this.$message({ message: '当前Server不是启动状态，无法停止', type: 'error' }) return } diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index e93956c9..368e4f59 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -32,7 +32,7 @@ com.alibaba.otter canal.client - 1.1.4-SNAPSHOT + ${project.version} org.yaml diff --git a/deployer/manager_ddl.sql b/deployer/manager_ddl.sql deleted file mode 100644 index c175c24e..00000000 --- a/deployer/manager_ddl.sql +++ /dev/null @@ -1,37 +0,0 @@ -CREATE SCHEMA `canal_manager` DEFAULT CHARACTER SET utf8mb4 ; - -CREATE TABLE `canal_config` ( - `id` bigint(20) NOT NULL AUTO_INCREMENT, - `name` varchar(45) NOT NULL, - `content` text NOT NULL, - `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`), - UNIQUE KEY `name_UNIQUE` (`name`) -) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 ; - -CREATE TABLE `canal_instance_config` ( - `id` bigint(20) NOT NULL AUTO_INCREMENT, - `name` varchar(45) NOT NULL, - `content` text NOT NULL, - `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`), - UNIQUE KEY `name_UNIQUE` (`name`) -) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 ; - -CREATE TABLE `canal_adapter_config` ( - `id` bigint(20) NOT NULL AUTO_INCREMENT, - `category` varchar(45) NOT NULL, - `name` varchar(45) NOT NULL, - `content` text NOT NULL, - `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (`id`) -) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 ; - -INSERT INTO `canal_config` VALUES ('1', 'canal.properties', '#################################################\r\n######### common argument ############# \r\n#################################################\r\n#canal.manager.jdbc.url=jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8\r\n#canal.manager.jdbc.username=root\r\n#canal.manager.jdbc.password=121212\r\ncanal.id = 1\r\ncanal.ip =\r\ncanal.port = 11111\r\ncanal.metrics.pull.port = 11112\r\ncanal.zkServers =\r\n# flush data to zk\r\ncanal.zookeeper.flush.period = 1000\r\ncanal.withoutNetty = false\r\n# tcp, kafka, RocketMQ\r\ncanal.serverMode = tcp\r\n# flush meta cursor/parse position to file\r\ncanal.file.data.dir = ${canal.conf.dir}\r\ncanal.file.flush.period = 1000\r\n## memory store RingBuffer size, should be Math.pow(2,n)\r\ncanal.instance.memory.buffer.size = 16384\r\n## memory store RingBuffer used memory unit size , default 1kb\r\ncanal.instance.memory.buffer.memunit = 1024 \r\n## meory store gets mode used MEMSIZE or ITEMSIZE\r\ncanal.instance.memory.batch.mode = MEMSIZE\r\ncanal.instance.memory.rawEntry = true\r\n\r\n## detecing config\r\ncanal.instance.detecting.enable = false\r\n#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()\r\ncanal.instance.detecting.sql = select 1\r\ncanal.instance.detecting.interval.time = 3\r\ncanal.instance.detecting.retry.threshold = 3\r\ncanal.instance.detecting.heartbeatHaEnable = false\r\n\r\n# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery\r\ncanal.instance.transaction.size = 1024\r\n# mysql fallback connected to new master should fallback times\r\ncanal.instance.fallbackIntervalInSeconds = 60\r\n\r\n# network config\r\ncanal.instance.network.receiveBufferSize = 16384\r\ncanal.instance.network.sendBufferSize = 16384\r\ncanal.instance.network.soTimeout = 30\r\n\r\n# binlog filter config\r\ncanal.instance.filter.druid.ddl = true\r\ncanal.instance.filter.query.dcl = false\r\ncanal.instance.filter.query.dml = false\r\ncanal.instance.filter.query.ddl = false\r\ncanal.instance.filter.table.error = false\r\ncanal.instance.filter.rows = false\r\ncanal.instance.filter.transaction.entry = false\r\n\r\n# binlog format/image check\r\ncanal.instance.binlog.format = ROW,STATEMENT,MIXED \r\ncanal.instance.binlog.image = FULL,MINIMAL,NOBLOB\r\n\r\n# binlog ddl isolation\r\ncanal.instance.get.ddl.isolation = false\r\n\r\n# parallel parser config\r\ncanal.instance.parser.parallel = true\r\n## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()\r\n#canal.instance.parser.parallelThreadSize = 16\r\n## disruptor ringbuffer size, must be power of 2\r\ncanal.instance.parser.parallelBufferSize = 256\r\n\r\n# table meta tsdb info\r\ncanal.instance.tsdb.enable = true\r\ncanal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}\r\ncanal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;\r\ncanal.instance.tsdb.dbUsername = canal\r\ncanal.instance.tsdb.dbPassword = canal\r\n# dump snapshot interval, default 24 hour\r\ncanal.instance.tsdb.snapshot.interval = 24\r\n# purge snapshot expire , default 360 hour(15 days)\r\ncanal.instance.tsdb.snapshot.expire = 360\r\n\r\n# aliyun ak/sk , support rds/mq\r\ncanal.aliyun.accesskey =\r\ncanal.aliyun.secretkey =\r\n\r\n#################################################\r\n######### destinations ############# \r\n#################################################\r\ncanal.destinations = example\r\n# conf root dir\r\ncanal.conf.dir = ../conf\r\n# auto scan instance dir add/remove and start/stop instance\r\ncanal.auto.scan = true\r\ncanal.auto.scan.interval = 5\r\n\r\ncanal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml\r\n#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml\r\n\r\ncanal.instance.global.mode = spring\r\ncanal.instance.global.lazy = false\r\n#canal.instance.global.manager.address = 127.0.0.1:1099\r\n#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml\r\ncanal.instance.global.spring.xml = classpath:spring/file-instance.xml\r\n#canal.instance.global.spring.xml = classpath:spring/default-instance.xml\r\n\r\n##################################################\r\n######### MQ #############\r\n##################################################\r\ncanal.mq.servers = 127.0.0.1:6667\r\ncanal.mq.retries = 0\r\ncanal.mq.batchSize = 16384\r\ncanal.mq.maxRequestSize = 1048576\r\ncanal.mq.lingerMs = 1\r\ncanal.mq.bufferMemory = 33554432\r\ncanal.mq.canalBatchSize = 50\r\ncanal.mq.canalGetTimeout = 100\r\ncanal.mq.flatMessage = true\r\ncanal.mq.compressionType = none\r\ncanal.mq.acks = all\r\n', '2018-12-30 16:46:00'); -INSERT INTO `canal_config` VALUES ('2', 'application.yml', 'server:\n port: 8081\nlogging:\n level:\n org.springframework: WARN\n com.alibaba.otter.canal.client.adapter.hbase: DEBUG\n com.alibaba.otter.canal.client.adapter.es: DEBUG\n com.alibaba.otter.canal.client.adapter.rdb: DEBUG\nspring:\n jackson:\n date-format: yyyy-MM-dd HH:mm:ss\n time-zone: GMT+8\n default-property-inclusion: non_null\n\ncanal.conf:\n canalServerHost: 127.0.0.1:11111\n# zookeeperHosts: slave1:2181\n# mqServers: 127.0.0.1:9092 #or rocketmq\n# flatMessage: true\n batchSize: 500\n syncBatchSize: 1000\n retries: 0\n timeout:\n accessKey:\n secretKey:\n mode: tcp # kafka rocketMQ\n# srcDataSources:\n# defaultDS:\n# url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true\n# username: root\n# password: 121212\n canalAdapters:\n - instance: example # canal instance Name or mq topic name\n groups:\n - groupId: g1\n outerAdapters:\n - name: logger\n# - name: rdb\n# key: mysql1\n# properties:\n# jdbc.driverClassName: com.mysql.jdbc.Driver\n# jdbc.url: jdbc:mysql://127.0.0.1:3306/mytest2?useUnicode=true\n# jdbc.username: root\n# jdbc.password: 121212\n# - name: rdb\n# key: oracle1\n# properties:\n# jdbc.driverClassName: oracle.jdbc.OracleDriver\n# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE\n# jdbc.username: mytest\n# jdbc.password: m121212\n# - name: rdb\n# key: postgres1\n# properties:\n# jdbc.driverClassName: org.postgresql.Driver\n# jdbc.url: jdbc:postgresql://localhost:5432/postgres\n# jdbc.username: postgres\n# jdbc.password: 121212\n# threads: 1\n# commitSize: 3000\n# - name: hbase\n# properties:\n# hbase.zookeeper.quorum: 127.0.0.1\n# hbase.zookeeper.property.clientPort: 2181\n# zookeeper.znode.parent: /hbase\n# - name: es\n# hosts: 127.0.0.1:9300\n# properties:\n# cluster.name: elasticsearch\n', '2019-01-05 01:37:43'); - -INSERT INTO `canal_instance_config` VALUES ('1', 'example', '#################################################\r\n## mysql serverId , v1.0.26+ will autoGen\r\n# canal.instance.mysql.slaveId=0\r\n\r\n# enable gtid use true/false\r\ncanal.instance.gtidon=false\r\n\r\n# position info\r\ncanal.instance.master.address=127.0.0.1:3306\r\ncanal.instance.master.journal.name=\r\ncanal.instance.master.position=\r\ncanal.instance.master.timestamp=\r\ncanal.instance.master.gtid=\r\n\r\n# rds oss binlog\r\ncanal.instance.rds.accesskey=\r\ncanal.instance.rds.secretkey=\r\ncanal.instance.rds.instanceId=\r\n\r\n# table meta tsdb info\r\ncanal.instance.tsdb.enable=true\r\n#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb\r\n#canal.instance.tsdb.dbUsername=canal\r\n#canal.instance.tsdb.dbPassword=canal\r\n\r\n#canal.instance.standby.address =\r\n#canal.instance.standby.journal.name =\r\n#canal.instance.standby.position =\r\n#canal.instance.standby.timestamp =\r\n#canal.instance.standby.gtid=\r\n\r\n# username/password\r\ncanal.instance.dbUsername=canal\r\ncanal.instance.dbPassword=canal\r\ncanal.instance.connectionCharset = UTF-8\r\ncanal.instance.defaultDatabaseName =test\r\n# enable druid Decrypt database password\r\ncanal.instance.enableDruid=false\r\n#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==\r\n\r\n# table regex\r\ncanal.instance.filter.regex=.*\\\\..*\r\n# table black regex\r\ncanal.instance.filter.black.regex=\r\n\r\n# mq config\r\ncanal.mq.topic=example\r\n# 动态topic, 需mq支持动态创建topic\r\n#canal.mq.dynamicTopic=.*,mytest\\\\..*,mytest2.user\r\ncanal.mq.partition=0\r\n# hash partition config\r\n#canal.mq.partitionsNum=3\r\n#canal.mq.partitionHash=test.table:id^name,.*\\\\..*\r\n#################################################\r\n', '2018-12-30 16:46:16'); - -INSERT INTO `canal_adapter_config` VALUES ('1', 'es', 'mytest_user.yml', 'dataSourceKey: defaultDS\ndestination: example\nesMapping:\n _index: mytest_user\n _type: _doc\n _id: _id\n# pk: id\n sql: \"select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,\n a.c_time as _c_time, c.labels as _labels from user a\n left join role b on b.id=a.role_id\n left join (select user_id, group_concat(label order by id desc separator \';\') as labels from label\n group by user_id) c on c.user_id=a.id\"\n# objFields:\n# _labels: array:;\n etlCondition: \"where a.c_time>=\'{0}\'\"\n commitBatch: 3000\n', '2019-01-01 17:05:43'); -INSERT INTO `canal_adapter_config` VALUES ('2', 'hbase', 'mytest_person2.yml', 'dataSourceKey: defaultDS\ndestination: example\nhbaseMapping:\n mode: STRING #NATIVE #PHOENIX\n database: mytest # 数据库名\n table: person2 # 数据库表名\n hbaseTable: MYTEST.PERSON2 # HBase表名\n family: CF # 默认统一Family名称\n uppercaseQualifier: true # 字段名转大写, 默认为true\n commitBatch: 3000 # 批量提交的大小\n #rowKey: id,type # 复合字段rowKey不能和columns中的rowKey重复\n columns:\n # 数据库字段:HBase对应字段\n id: ROWKEY LEN:15\n name: NAME\n email: EMAIL\n type:\n c_time: C_TIME\n birthday: BIRTHDAY\n# excludeColumns:\n# - lat # 忽略字段\n\n# -- NATIVE类型\n# $DEFAULT\n# $STRING\n# $INTEGER\n# $LONG\n# $SHORT\n# $BOOLEAN\n# $FLOAT\n# $DOUBLE\n# $BIGDECIMAL\n# $DATE\n# $BYTE\n# $BYTES\n\n# -- PHOENIX类型\n# $DEFAULT 对应PHOENIX里的VARCHAR\n# $UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节\n# $UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节\n# $UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节\n# $UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节\n# $UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节\n# $UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节\n# $INTEGER 对应PHOENIX里的INTEGER 4字节\n# $BIGINT 对应PHOENIX里的BIGINT 8字节\n# $TINYINT 对应PHOENIX里的TINYINT 1字节\n# $SMALLINT 对应PHOENIX里的SMALLINT 2字节\n# $FLOAT 对应PHOENIX里的FLOAT 4字节\n# $DOUBLE 对应PHOENIX里的DOUBLE 8字节\n# $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节\n# $TIME 对应PHOENIX里的TIME 8字节\n# $DATE 对应PHOENIX里的DATE 8字节\n# $TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节\n# $UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节\n# $UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节\n# $UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节\n# $VARCHAR 对应PHOENIX里的VARCHAR 动态长度\n# $VARBINARY 对应PHOENIX里的VARBINARY 动态长度\n# $DECIMAL 对应PHOENIX里的DECIMAL 动态长度', '2019-01-01 17:08:08'); -INSERT INTO `canal_adapter_config` VALUES ('3', 'rdb', 'mytest_user.yml', 'dataSourceKey: defaultDS\ndestination: example\nouterAdapterKey: oracle1\nconcurrent: true\ndbMapping:\n database: mytest\n table: user\n targetTable: mytest.tb_user\n targetPk:\n id: id\n# mapAll: true\n targetColumns:\n id:\n name:\n role_id:\n c_time:\n test1:\n\n\n# Mirror schema synchronize config\n#dataSourceKey: defaultDS\n#destination: example\n#outerAdapterKey: mysql1\n#concurrent: true\n#dbMapping:\n# mirrorDb: true\n# database: mytest\n', '2019-01-01 17:08:50'); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index b36188b3..93534b02 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -19,6 +19,7 @@ public class CanalConstants { public static final String CANAL_USER = ROOT + "." + "user"; public static final String CANAL_PASSWD = ROOT + "." + "passwd"; public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port"; + public static final String CANAL_ADMIN_MANAGER = ROOT + "." + "admin.manager"; public static final String CANAL_ADMIN_PORT = ROOT + "." + "admin.port"; public static final String CANAL_ADMIN_USER = ROOT + "." + "admin.user"; public static final String CANAL_ADMIN_PASSWD = ROOT + "." + "admin.passwd"; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index 537a58d6..24ac6545 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -12,11 +12,7 @@ import org.apache.zookeeper.Watcher.Event.KeeperState; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.context.ApplicationContext; -import org.springframework.context.support.ClassPathXmlApplicationContext; -import com.alibaba.otter.canal.common.CanalException; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.common.zookeeper.ZkClientx; import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; @@ -31,10 +27,9 @@ import com.alibaba.otter.canal.deployer.monitor.ManagerInstanceConfigMonitor; import com.alibaba.otter.canal.deployer.monitor.SpringInstanceConfigMonitor; import com.alibaba.otter.canal.instance.core.CanalInstance; import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator; -import com.alibaba.otter.canal.instance.manager.CanalConfigClient; -import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator; +import com.alibaba.otter.canal.instance.manager.PlainCanalInstanceGenerator; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient; import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator; -import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.server.CanalMQStarter; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; import com.alibaba.otter.canal.server.exception.CanalServerException; @@ -56,10 +51,11 @@ public class CanalController { private String ip; private String registerIp; private int port; + private int adminPort; // 默认使用spring的方式载入 private Map instanceConfigs; private InstanceConfig globalInstanceConfig; - private Map managerClients; + private Map managerClients; // 监听instance config的变化 private boolean autoScan = true; private InstanceAction defaultAction; @@ -71,15 +67,17 @@ public class CanalController { private ZkClientx zkclientx; private CanalMQStarter canalMQStarter; + private String adminUser; + private String adminPasswd; public CanalController(){ this(System.getProperties()); } public CanalController(final Properties properties){ - managerClients = MigrateMap.makeComputingMap(new Function() { + managerClients = MigrateMap.makeComputingMap(new Function() { - public CanalConfigClient apply(String managerAddress) { + public PlainCanalConfigClient apply(String managerAddress) { return getManagerClient(managerAddress); } }); @@ -111,6 +109,7 @@ public class CanalController { ip = getProperty(properties, CanalConstants.CANAL_IP); registerIp = getProperty(properties, CanalConstants.CANAL_REGISTER_IP); port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT)); + adminPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_ADMIN_PORT)); embededCanalServer = CanalServerWithEmbedded.instance(); embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator try { @@ -121,6 +120,8 @@ public class CanalController { embededCanalServer.setMetricsPort(11112); } + this.adminUser = getProperty(properties, CanalConstants.CANAL_ADMIN_USER); + this.adminPasswd = getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD); embededCanalServer.setUser(getProperty(properties, CanalConstants.CANAL_USER)); embededCanalServer.setPasswd(getProperty(properties, CanalConstants.CANAL_PASSWD)); @@ -254,6 +255,8 @@ public class CanalController { runningMonitor.start(); } } + + logger.info("auto notify start {} successful.", destination); } public void stop(String destination) { @@ -266,12 +269,16 @@ public class CanalController { runningMonitor.stop(); } } + + logger.info("auto notify stop {} successful.", destination); } public void reload(String destination) { // 目前任何配置变化，直接重启，简单处理 stop(destination); start(destination); + + logger.info("auto notify reload {} successful.", destination); } }; @@ -298,7 +305,14 @@ public class CanalController { } return monitor; } else if (mode.isManager()) { - return new ManagerInstanceConfigMonitor(); + ManagerInstanceConfigMonitor monitor = new ManagerInstanceConfigMonitor(); + monitor.setScanIntervalInSecond(scanInterval); + monitor.setDefaultAction(defaultAction); + String managerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER); + monitor.setConfigClient(getManagerClient(managerAddress)); + monitor.setIp(registerIp); + monitor.setPort(adminPort); + return monitor; } else { throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor"); } @@ -308,9 +322,13 @@ public class CanalController { } private InstanceConfig initGlobalConfig(Properties properties) { + String adminManagerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER); InstanceConfig globalConfig = new InstanceConfig(); String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(CanalConstants.GLOBAL_NAME)); - if (StringUtils.isNotEmpty(modeStr)) { + if (StringUtils.isNotEmpty(adminManagerAddress)) { + // 如果指定了manager地址,则强制适用manager + globalConfig.setMode(InstanceMode.MANAGER); + } else if (StringUtils.isNotEmpty(modeStr)) { globalConfig.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr))); } @@ -322,6 +340,10 @@ public class CanalController { String managerAddress = getProperty(properties, CanalConstants.getInstanceManagerAddressKey(CanalConstants.GLOBAL_NAME)); if (StringUtils.isNotEmpty(managerAddress)) { + if (StringUtils.equals(managerAddress, "${canal.admin.manager}")) { + managerAddress = adminManagerAddress; + } + globalConfig.setManagerAddress(managerAddress); } @@ -335,28 +357,18 @@ public class CanalController { public CanalInstance generate(String destination) { InstanceConfig config = instanceConfigs.get(destination); if (config == null) { - throw new CanalServerException("can't find destination:{}"); + throw new CanalServerException("can't find destination:" + destination); } if (config.getMode().isManager()) { - ManagerCanalInstanceGenerator instanceGenerator = new ManagerCanalInstanceGenerator(); + PlainCanalInstanceGenerator instanceGenerator = new PlainCanalInstanceGenerator(); instanceGenerator.setCanalConfigClient(managerClients.get(config.getManagerAddress())); + instanceGenerator.setSpringXml(config.getSpringXml()); return instanceGenerator.generate(destination); } else if (config.getMode().isSpring()) { SpringCanalInstanceGenerator instanceGenerator = new SpringCanalInstanceGenerator(); - synchronized (CanalEventParser.class) { - try { - // 设置当前正在加载的通道，加载spring查找文件时会用到该变量 - System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, destination); - instanceGenerator.setBeanFactory(getBeanFactory(config.getSpringXml())); - return instanceGenerator.generate(destination); - } catch (Throwable e) { - logger.error("generator instance failed.", e); - throw new CanalException(e); - } finally { - System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, ""); - } - } + instanceGenerator.setSpringXml(config.getSpringXml()); + return instanceGenerator.generate(destination); } else { throw new UnsupportedOperationException("unknow mode :" + config.getMode()); } @@ -368,13 +380,8 @@ public class CanalController { return globalConfig; } - private CanalConfigClient getManagerClient(String managerAddress) { - return new CanalConfigClient(); - } - - private BeanFactory getBeanFactory(String springXml) { - ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml); - return applicationContext; + private PlainCanalConfigClient getManagerClient(String managerAddress) { + return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd); } private void initInstanceConfig(Properties properties) { @@ -392,9 +399,13 @@ public class CanalController { } private InstanceConfig parseInstanceConfig(Properties properties, String destination) { + String adminManagerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER); InstanceConfig config = new InstanceConfig(globalInstanceConfig); String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(destination)); - if (!StringUtils.isEmpty(modeStr)) { + if (StringUtils.isNotEmpty(adminManagerAddress)) { + // 如果指定了manager地址,则强制适用manager + config.setMode(InstanceMode.MANAGER); + } else if (StringUtils.isNotEmpty(modeStr)) { config.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr))); } @@ -406,6 +417,9 @@ public class CanalController { if (config.getMode().isManager()) { String managerAddress = getProperty(properties, CanalConstants.getInstanceManagerAddressKey(destination)); if (StringUtils.isNotEmpty(managerAddress)) { + if (StringUtils.equals(managerAddress, "${canal.admin.manager}")) { + managerAddress = adminManagerAddress; + } config.setManagerAddress(managerAddress); } } else if (config.getMode().isSpring()) { diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 8ce96dd4..4d0cf900 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -3,14 +3,17 @@ package com.alibaba.otter.canal.deployer; import java.io.FileInputStream; import java.util.Properties; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.deployer.monitor.remote.RemoteCanalConfigMonitor; -import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoader; -import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoaderFactory; +import com.alibaba.otter.canal.common.utils.NamedThreadFactory; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanal; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient; /** * canal独立版本启动的入口类 @@ -20,9 +23,11 @@ import com.alibaba.otter.canal.deployer.monitor.remote.RemoteConfigLoaderFactory */ public class CanalLauncher { - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); - public static final CountDownLatch runningLatch = new CountDownLatch(1); + private static final String CLASSPATH_URL_PREFIX = "classpath:"; + private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + public static final CountDownLatch runningLatch = new CountDownLatch(1); + private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, + new NamedThreadFactory("canal-server-scan")); public static void main(String[] args) { try { @@ -32,7 +37,6 @@ public class CanalLauncher { logger.info("## load canal configurations"); String conf = System.getProperty("canal.conf", "classpath:canal.properties"); Properties properties = new Properties(); - RemoteConfigLoader remoteConfigLoader = null; if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf)); @@ -40,43 +44,48 @@ public class CanalLauncher { properties.load(new FileInputStream(conf)); } - remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(properties); - if (remoteConfigLoader != null) { - // 加载远程canal.properties - Properties remoteConfig = remoteConfigLoader.loadRemoteConfig(); - // 加载remote instance配置 - remoteConfigLoader.loadRemoteInstanceConfigs(); - if (remoteConfig != null) { - properties = remoteConfig; - } else { - remoteConfigLoader = null; - } - } - final CanalStater canalStater = new CanalStater(properties); - canalStater.start(); + String managerAddress = properties.getProperty(CanalConstants.CANAL_ADMIN_MANAGER); + if (StringUtils.isNotEmpty(managerAddress)) { + String user = properties.getProperty(CanalConstants.CANAL_ADMIN_USER); + String passwd = properties.getProperty(CanalConstants.CANAL_ADMIN_PASSWD); + final PlainCanalConfigClient configClient = new PlainCanalConfigClient(managerAddress, user, passwd); + PlainCanal canalConfig = configClient.findServer(null); + properties = canalConfig.getProperties(); + int scanIntervalInSecond = Integer.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, + "5")); + executor.scheduleWithFixedDelay(new Runnable() { - if (remoteConfigLoader != null) { - remoteConfigLoader.startMonitor(new RemoteCanalConfigMonitor() { + private PlainCanal lastCanalConfig; - @Override - public void onChange(Properties properties) { + public void run() { try { - // 远程配置canal.properties修改重新加载整个应用 - canalStater.stop(); - canalStater.setProperties(properties); - canalStater.start(); - } catch (Throwable throwable) { - logger.error(throwable.getMessage(), throwable); + if (lastCanalConfig == null) { + lastCanalConfig = configClient.findServer(null); + } else { + PlainCanal newCanalConfig = configClient.findServer(lastCanalConfig.getMd5()); + if (newCanalConfig != null) { + // 远程配置canal.properties修改重新加载整个应用 + canalStater.stop(); + canalStater.setProperties(newCanalConfig.getProperties()); + canalStater.start(); + + lastCanalConfig = newCanalConfig; + } + } + + } catch (Throwable e) { + logger.error("scan failed", e); } } - }); + + }, 0, scanIntervalInSecond, TimeUnit.SECONDS); } + canalStater.setProperties(properties); + canalStater.start(); runningLatch.await(); - if (remoteConfigLoader != null) { - remoteConfigLoader.destroy(); - } + executor.shutdownNow(); } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java index 0844ad10..8c356e9d 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java @@ -19,6 +19,7 @@ import com.alibaba.otter.canal.deployer.CanalStater; import com.alibaba.otter.canal.deployer.InstanceConfig; import com.alibaba.otter.canal.deployer.monitor.InstanceAction; import com.alibaba.otter.canal.deployer.monitor.InstanceConfigMonitor; +import com.alibaba.otter.canal.deployer.monitor.ManagerInstanceConfigMonitor; import com.alibaba.otter.canal.deployer.monitor.SpringInstanceConfigMonitor; import com.alibaba.otter.canal.instance.core.CanalInstance; import com.alibaba.otter.canal.protocol.SecurityUtil; @@ -204,9 +205,24 @@ public class CanalAdminController implements CanalAdmin { private InstanceAction getInstanceAction(String destination) { Map monitors = canalStater.getController() .getInstanceConfigMonitors(); - SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.SPRING); - Map instanceActions = monitor.getActions(); - return instanceActions.get(destination); + + InstanceAction instanceAction = null; + if (monitors.containsKey(InstanceConfig.InstanceMode.SPRING)) { + SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.SPRING); + Map instanceActions = monitor.getActions(); + instanceAction = instanceActions.get(destination); + } + + if (instanceAction != null) { + return instanceAction; + } + + if (monitors.containsKey(InstanceConfig.InstanceMode.MANAGER)) { + ManagerInstanceConfigMonitor monitor = (ManagerInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.MANAGER); + Map instanceActions = monitor.getActions(); + instanceAction = instanceActions.get(destination); + } + return instanceAction; } public void setUser(String user) { diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java index d6099f29..4430bcf3 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java @@ -1,20 +1,185 @@ package com.alibaba.otter.canal.deployer.monitor; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.CanalLifeCycle; +import com.alibaba.otter.canal.common.utils.NamedThreadFactory; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanal; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient; +import com.google.common.base.Function; +import com.google.common.collect.Lists; +import com.google.common.collect.MapMaker; +import com.google.common.collect.MigrateMap; /** - * @author jianghang 2013-2-18 下午03:19:06 - * @version 1.0.1 + * 基于manager配置的实现 + * + * @author agapple 2019年8月26日 下午10:00:20 + * @since 1.1.4 */ public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle { - public void register(String destination, InstanceAction action) { + private static final Logger logger = LoggerFactory.getLogger(ManagerInstanceConfigMonitor.class); + private long scanIntervalInSecond = 5; + private InstanceAction defaultAction = null; + private Map actions = new MapMaker().makeMap(); + private Map configs = MigrateMap.makeComputingMap(new Function() { + public PlainCanal apply(String destination) { + return new PlainCanal(); + } + }); + private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, + new NamedThreadFactory("canal-instance-scan")); + + private volatile boolean isFirst = true; + private PlainCanalConfigClient configClient; + private String ip; + private int port; + private String lastInstanceMD5; + + public void start() { + super.start(); + executor.scheduleWithFixedDelay(new Runnable() { + + public void run() { + try { + scan(); + if (isFirst) { + isFirst = false; + } + } catch (Throwable e) { + logger.error("scan failed", e); + } + } + + }, 0, scanIntervalInSecond, TimeUnit.SECONDS); + } + + public void stop() { + super.stop(); + executor.shutdownNow(); + actions.clear(); + } + + public void register(String destination, InstanceAction action) { + if (action != null) { + actions.put(destination, action); + } else { + actions.put(destination, defaultAction); + } } public void unregister(String destination) { + actions.remove(destination); + } + private void scan() { + String instances = configClient.findInstances(ip, String.valueOf(port), lastInstanceMD5); + final List is = Lists.newArrayList(StringUtils.split(instances, ',')); + List start = Lists.newArrayList(); + List stop = Lists.newArrayList(); + List restart = Lists.newArrayList(); + for (String instance : is) { + if (!configs.containsKey(instance)) { + PlainCanal newPlainCanal = configClient.findInstance(instance, null); + if (newPlainCanal != null) { + configs.put(instance, newPlainCanal); + start.add(instance); + } + } else { + PlainCanal plainCanal = configs.get(instance); + PlainCanal newPlainCanal = configClient.findInstance(instance, plainCanal.getMd5()); + if (newPlainCanal != null) { + // 配置有变化 + restart.add(instance); + configs.put(instance, newPlainCanal); + } + } + } + + configs.forEach((instance, plainCanal) -> { + if (!is.contains(instance)) { + stop.add(instance); + } + }); + + stop.forEach(instance -> { + notifyStop(instance); + configs.remove(instance); + }); + + restart.forEach(instance -> { + notifyReload(instance); + }); + + start.forEach(instance -> { + notifyStart(instance); + }); + } + + private void notifyStart(String destination) { + try { + defaultAction.start(destination); + actions.put(destination, defaultAction); + // 启动成功后记录配置文件信息 + } catch (Throwable e) { + logger.error(String.format("scan add found[%s] but start failed", destination), e); + } + } + + private void notifyStop(String destination) { + InstanceAction action = actions.remove(destination); + try { + action.stop(destination); + } catch (Throwable e) { + logger.error(String.format("scan delete found[%s] but stop failed", destination), e); + actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除 + } + } + + private void notifyReload(String destination) { + InstanceAction action = actions.get(destination); + if (action != null) { + try { + action.reload(destination); + } catch (Throwable e) { + logger.error(String.format("scan reload found[%s] but reload failed", destination), e); + } + } + } + + public void setDefaultAction(InstanceAction defaultAction) { + this.defaultAction = defaultAction; + } + + public void setScanIntervalInSecond(long scanIntervalInSecond) { + this.scanIntervalInSecond = scanIntervalInSecond; + } + + public void setConfigClient(PlainCanalConfigClient configClient) { + this.configClient = configClient; + } + + public void setIp(String ip) { + this.ip = ip; + } + + public void setPort(int port) { + this.port = port; + } + + public Map getActions() { + return actions; } } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java index b8cef713..5b2e6805 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java @@ -186,8 +186,6 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen newFileInfo.add(new FileInfo(instanceConfig.getName(), instanceConfig.lastModified())); } lastFile.setInstanceFiles(newFileInfo); - - logger.info("auto notify start {} successful.", destination); } catch (Throwable e) { logger.error(String.format("scan add found[%s] but start failed", destination), e); } @@ -198,7 +196,6 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen try { action.stop(destination); lastFiles.remove(destination); - logger.info("auto notify stop {} successful.", destination); } catch (Throwable e) { logger.error(String.format("scan delete found[%s] but stop failed", destination), e); actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除 @@ -210,7 +207,6 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen if (action != null) { try { action.reload(destination); - logger.info("auto notify reload {} successful.", destination); } catch (Throwable e) { logger.error(String.format("scan reload found[%s] but reload failed", destination), e); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java deleted file mode 100644 index 5fb9b4c6..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/ConfigItem.java +++ /dev/null @@ -1,47 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -/** - * 配置对应对象 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public class ConfigItem { - - private Long id; - private String name; - private String content; - private long modifiedTime; - - public Long getId() { - return id; - } - - public void setId(Long id) { - this.id = id; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public String getContent() { - return content; - } - - public void setContent(String content) { - this.content = content; - } - - public long getModifiedTime() { - return modifiedTime; - } - - public void setModifiedTime(long modifiedTime) { - this.modifiedTime = modifiedTime; - } -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java deleted file mode 100644 index 1ef74d88..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/DbRemoteConfigLoader.java +++ /dev/null @@ -1,264 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -import java.io.ByteArrayInputStream; -import java.io.FileWriter; -import java.nio.charset.StandardCharsets; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.*; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.druid.pool.DruidDataSource; -import com.alibaba.otter.canal.common.utils.CommonUtils; -import com.alibaba.otter.canal.common.utils.NamedThreadFactory; -import com.alibaba.otter.canal.deployer.CanalConstants; -import com.google.common.base.Joiner; -import com.google.common.collect.MapMaker; - -/** - * 基于数据库的远程配置装载器 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public class DbRemoteConfigLoader implements RemoteConfigLoader { - - private static final Logger logger = LoggerFactory.getLogger(DbRemoteConfigLoader.class); - - private Map remoteInstanceConfigs = new MapMaker().makeMap(); - - private DruidDataSource dataSource; - - private long currentConfigTimestamp = 0; - - private long scanIntervalInSecond = 5; - private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, - new NamedThreadFactory("remote-canal-config-scan")); - - private RemoteInstanceMonitor remoteInstanceMonitor = new RemoteInstanceMonitorImpl(); - - public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){ - dataSource = new DruidDataSource(); - if (StringUtils.isEmpty(driverName)) { - driverName = "com.mysql.jdbc.Driver"; - } - dataSource.setDriverClassName(driverName); - dataSource.setUrl(jdbcUrl); - dataSource.setUsername(jdbcUsername); - dataSource.setPassword(jdbcPassword); - dataSource.setInitialSize(1); - dataSource.setMinIdle(1); - dataSource.setMaxActive(1); - dataSource.setMaxWait(60000); - dataSource.setTimeBetweenEvictionRunsMillis(60000); - dataSource.setMinEvictableIdleTimeMillis(300000); - try { - dataSource.init(); - } catch (SQLException e) { - throw new RuntimeException(e.getMessage(), e); - } - } - - /** - * 加载远程 canal.properties文件 - * - * @return 远程配置的properties - */ - @Override - public Properties loadRemoteConfig() { - Properties properties = null; - try { - // 加载远程canal配置 - ConfigItem configItem = getRemoteCanalConfig(); - if (configItem != null) { - if (configItem.getModifiedTime() != currentConfigTimestamp) { - currentConfigTimestamp = configItem.getModifiedTime(); - overrideLocalCanalConfig(configItem.getContent()); - properties = new Properties(); - properties.load(new ByteArrayInputStream(configItem.getContent().getBytes(StandardCharsets.UTF_8))); - scanIntervalInSecond = Integer - .parseInt(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5")); - logger.info("## Loaded remote canal config: canal.properties "); - } - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - return properties; - } - - /** - * 覆盖本地 canal.properties - * - * @param content 远程配置内容文本 - */ - private void overrideLocalCanalConfig(String content) { - try (FileWriter writer = new FileWriter(CommonUtils.getConfPath() + "canal.properties")) { - writer.write(content); - writer.flush(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - /** - * 获取远程canal.properties配置内容 - * - * @return 内容对象 - */ - private ConfigItem getRemoteCanalConfig() { - String sql = "select name, content, modified_time from canal_config where id=1"; - try (Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(sql)) { - if (rs.next()) { - ConfigItem configItem = new ConfigItem(); - configItem.setId(1L); - configItem.setName(rs.getString("name")); - configItem.setContent(rs.getString("content")); - configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); - return configItem; - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - return null; - } - - /** - * 加载远程的instance配置 - */ - @Override - public void loadRemoteInstanceConfigs() { - try { - // 加载远程instance配置 - loadModifiedInstanceConfigs(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - /** - * 加载远程instance新增、修改、删除配置 - */ - private void loadModifiedInstanceConfigs() { - Map remoteConfigStatus = new HashMap<>(); - String sql = "select id, name, modified_time from canal_instance_config"; - try (Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(sql)) { - while (rs.next()) { - ConfigItem configItem = new ConfigItem(); - configItem.setId(rs.getLong("id")); - configItem.setName(rs.getString("name")); - configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); - remoteConfigStatus.put(configItem.getName(), configItem); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - - if (!remoteConfigStatus.isEmpty()) { - List changedIds = new ArrayList<>(); - - for (ConfigItem remoteConfigStat : remoteConfigStatus.values()) { - ConfigItem currentConfig = remoteInstanceConfigs.get(remoteConfigStat.getName()); - if (currentConfig == null) { - // 新增 - changedIds.add(remoteConfigStat.getId()); - } else { - // 修改 - if (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { - changedIds.add(remoteConfigStat.getId()); - } - } - } - if (!changedIds.isEmpty()) { - String contentsSql = "select id, name, content, modified_time from canal_instance_config where id in (" - + Joiner.on(",").join(changedIds) + ")"; - try (Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(contentsSql)) { - while (rs.next()) { - ConfigItem configItemNew = new ConfigItem(); - configItemNew.setId(rs.getLong("id")); - configItemNew.setName(rs.getString("name")); - configItemNew.setContent(rs.getString("content")); - configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); - - remoteInstanceConfigs.put(configItemNew.getName(), configItemNew); - remoteInstanceMonitor.onModify(configItemNew); - } - - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - } - - for (String name : remoteInstanceConfigs.keySet()) { - if (!remoteConfigStatus.containsKey(name)) { - // 删除 - remoteInstanceConfigs.remove(name); - remoteInstanceMonitor.onDelete(name); - } - } - } - - /** - * 监听 canal 主配置和 instance 配置变化 - * - * @param remoteCanalConfigMonitor 监听回调方法 - */ - public void startMonitor(final RemoteCanalConfigMonitor remoteCanalConfigMonitor) { - // 监听canal.properties变化 - executor.scheduleWithFixedDelay(new Runnable() { - - public void run() { - try { - Properties properties = loadRemoteConfig(); - if (properties != null) { - remoteCanalConfigMonitor.onChange(properties); - } - } catch (Throwable e) { - logger.error("Scan remote canal config failed", e); - } - } - - }, 10, scanIntervalInSecond, TimeUnit.SECONDS); - - // 监听instance变化 - executor.scheduleWithFixedDelay(new Runnable() { - - public void run() { - try { - loadRemoteInstanceConfigs(); - } catch (Throwable e) { - logger.error("Scan remote instance config failed", e); - } - } - - }, 10, 3, TimeUnit.SECONDS); - } - - /** - * 销毁 - */ - public void destroy() { - executor.shutdownNow(); - try { - dataSource.close(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java deleted file mode 100644 index 4c0ab955..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteCanalConfigMonitor.java +++ /dev/null @@ -1,13 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -import java.util.Properties; - -/** - * 远程canal.properties配置监听器接口 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public interface RemoteCanalConfigMonitor { - void onChange(Properties properties); -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java deleted file mode 100644 index 6b7444c0..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoader.java +++ /dev/null @@ -1,36 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -import java.util.Properties; - -/** - * 远程配置装载器接口 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public interface RemoteConfigLoader { - - /** - * 加载远程 canal.properties文件 - * - * @return 远程配置的properties - */ - Properties loadRemoteConfig(); - - /** - * 加载远程的instance配置 - */ - void loadRemoteInstanceConfigs(); - - /** - * 启动监听 canal 主配置和 instance 配置变化 - * - * @param remoteCanalConfigMonitor 监听回调方法 - */ - void startMonitor(final RemoteCanalConfigMonitor remoteCanalConfigMonitor); - - /** - * 销毁 - */ - void destroy(); -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java deleted file mode 100644 index 87bde803..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteConfigLoaderFactory.java +++ /dev/null @@ -1,42 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -import java.util.Properties; - -import com.alibaba.otter.canal.deployer.monitor.remote.http.HttpRemoteConfigLoader; -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * 远程配置装载器工厂类 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public class RemoteConfigLoaderFactory { - - private static final Logger logger = LoggerFactory.getLogger(RemoteConfigLoaderFactory.class); - - public static RemoteConfigLoader getRemoteConfigLoader(Properties localProperties) { - String jdbcUrl = localProperties.getProperty("canal.manager.jdbc.url"); - String httpUrl = localProperties.getProperty("canal.manager.http.url"); - if (!StringUtils.isEmpty(jdbcUrl)) { - logger.info("## load remote db canal configurations"); - // load remote config - String driverName = localProperties.getProperty("canal.manager.jdbc.driverName"); - String jdbcUsername = localProperties.getProperty("canal.manager.jdbc.username"); - String jdbcPassword = localProperties.getProperty("canal.manager.jdbc.password"); - return new DbRemoteConfigLoader(driverName, jdbcUrl, jdbcUsername, jdbcPassword); - } else if (!StringUtils.isEmpty(httpUrl)) { - logger.info("## load remote http canal configurations"); - String httpUsername = localProperties.getProperty("canal.manager.http.username"); - String httpPassword = localProperties.getProperty("canal.manager.http.password"); - return new HttpRemoteConfigLoader(httpUrl, httpUsername, httpPassword); - } - // 可扩展其它远程配置加载器 - - logger.info("## load local canal configurations"); - - return null; - } -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java deleted file mode 100644 index 28c97788..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitor.java +++ /dev/null @@ -1,31 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -/** - * 远程xxx/instance.properties配置监听器接口 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public interface RemoteInstanceMonitor { - - /** - * 新增配置事件 - * - * @param configItem 配置项 - */ - void onAdd(ConfigItem configItem); - - /** - * 修改配置事件 - * - * @param configItem 配置项 - */ - void onModify(ConfigItem configItem); - - /** - * 删除配置事件 - * - * @param instanceName 实例名 - */ - void onDelete(String instanceName); -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitorImpl.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitorImpl.java deleted file mode 100644 index 8eded2b1..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/RemoteInstanceMonitorImpl.java +++ /dev/null @@ -1,55 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote; - -import java.io.File; -import java.io.FileWriter; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.otter.canal.common.utils.CommonUtils; - -/** - * 远程xxx/instance.properties配置监听器实现 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public class RemoteInstanceMonitorImpl implements RemoteInstanceMonitor { - - private static final Logger logger = LoggerFactory.getLogger(RemoteInstanceMonitorImpl.class); - - @Override - public void onAdd(ConfigItem configItem) { - this.onModify(configItem); - } - - @Override - public void onModify(ConfigItem configItem) { - String confDir = CommonUtils.getConfPath() + configItem.getName(); - File instanceDir = new File(confDir); - if (!instanceDir.exists()) { - boolean mkDirs = instanceDir.mkdirs(); - if (!mkDirs) { - logger.info("## Error to create instance config dir: {}", configItem.getName()); - return; - } - } - try (FileWriter writer = new FileWriter(confDir + "/instance.properties")) { - writer.write(configItem.getContent()); - writer.flush(); - logger.info("## Loaded remote instance config: {}/instance.properties ", configItem.getName()); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - @Override - public void onDelete(String instanceName) { - File file = new File(CommonUtils.getConfPath() + instanceName + "/"); - if (file.exists()) { - CommonUtils.deleteDir(file); - logger.info("## Deleted and loaded remote instance config: {} ", instanceName); - } - } - -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpRemoteConfigLoader.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpRemoteConfigLoader.java deleted file mode 100644 index 3d47e077..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpRemoteConfigLoader.java +++ /dev/null @@ -1,257 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote.http; - -import com.alibaba.fastjson.JSONObject; -import com.alibaba.fastjson.TypeReference; -import com.alibaba.otter.canal.common.utils.CommonUtils; -import com.alibaba.otter.canal.common.utils.NamedThreadFactory; -import com.alibaba.otter.canal.deployer.CanalConstants; -import com.alibaba.otter.canal.deployer.monitor.remote.*; -import com.google.common.collect.MapMaker; -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.ByteArrayInputStream; -import java.io.FileWriter; -import java.nio.charset.StandardCharsets; -import java.util.*; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; - -/** - * 基于HTTP的远程配置装载器 - * - * @author rewerma 2019-08-26 上午09:40:36 - * @version 1.0.0 - */ -public class HttpRemoteConfigLoader implements RemoteConfigLoader { - - private final static Logger logger = LoggerFactory.getLogger(HttpRemoteConfigLoader.class); - - private final static Integer REQUEST_TIMEOUT = 5000; - - private String baseUrl; - private String username; - private String password; - - private String token; - - private HttpHelper httpHelper; - - private long currentConfigTimestamp = 0; - private Map remoteInstanceConfigs = new MapMaker().makeMap(); - - private RemoteInstanceMonitor remoteInstanceMonitor = new RemoteInstanceMonitorImpl(); - - private long scanIntervalInSecond = 5; - private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, - new NamedThreadFactory("remote-http-canal-config-scan")); - - public HttpRemoteConfigLoader(String baseUrl, String username, String password){ - this.baseUrl = baseUrl; - this.username = username; - this.password = password; - httpHelper = new HttpHelper(); - } - - private String login4Token(HttpHelper httpHelper) { - Map reqBody = new HashMap<>(); - reqBody.put("username", username); - reqBody.put("password", password); - String response = httpHelper.post(baseUrl + "/api/v1/user/login", null, reqBody, REQUEST_TIMEOUT); - ResponseModel resp = JSONObject.parseObject(response, - new TypeReference>() { - }); - if (!HttpHelper.REST_STATE_OK.equals(resp.getCode())) { - throw new RuntimeException("requestPost for login error: " + resp.getMessage()); - } - return (String) resp.getData().get("token"); - } - - @Override - public synchronized Properties loadRemoteConfig() { - Properties properties = null; - try { - ConfigItem configItem = getRemoteCanalConfig(); - if (configItem != null) { - if (configItem.getModifiedTime() != currentConfigTimestamp) { // 修改时就不同说明配置有更新 - Map heads = new HashMap<>(); - heads.put("X-Token", token); - String response = httpHelper.get(baseUrl + "/api/v1/canal/config", heads, REQUEST_TIMEOUT); - ResponseModel resp = JSONObject.parseObject(response, - new TypeReference>() { - }); - currentConfigTimestamp = configItem.getModifiedTime(); - overrideLocalCanalConfig(resp.getData().getContent()); - properties = new Properties(); - properties - .load(new ByteArrayInputStream(resp.getData().getContent().getBytes(StandardCharsets.UTF_8))); - scanIntervalInSecond = Integer - .parseInt(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5")); - logger.info("## Loaded http remote canal config: canal.properties "); - } - } - } catch (Throwable t) { - logger.error(t.getMessage(), t); - } - return properties; - } - - /** - * 获取远程canal server配置摘要(无配置内容) - * - * @return 配置摘要 - */ - private ConfigItem getRemoteCanalConfig() { - if (StringUtils.isEmpty(token)) { - token = login4Token(httpHelper); - } - - Map heads = new HashMap<>(); - heads.put("X-Token", token); - String response = httpHelper.get(baseUrl + "/api/v1/canal/config/summary", heads, REQUEST_TIMEOUT); - ResponseModel resp = JSONObject.parseObject(response, - new TypeReference>() { - }); - if (HttpHelper.REST_STATE_TOKEN_INVALID.equals(resp.getCode())) { - // token 失效 - token = login4Token(httpHelper); - heads.put("X-Token", token); - response = httpHelper.get(baseUrl + "/api/v1/canal/config/summary", heads, REQUEST_TIMEOUT); - resp = JSONObject.parseObject(response, new TypeReference>() { - }); - } - if (!HttpHelper.REST_STATE_OK.equals(resp.getCode())) { - throw new RuntimeException("requestGet for canal config error: " + resp.getMessage()); - } - return resp.getData(); - } - - @Override - public synchronized void loadRemoteInstanceConfigs() { - try { - // 加载远程instance配置 - loadModifiedInstanceConfigs(); - } catch (Throwable t) { - logger.error(t.getMessage(), t); - } - } - - /** - * 加载远程instance新增、修改、删除配置 - */ - private void loadModifiedInstanceConfigs() { - if (StringUtils.isEmpty(token)) { - token = login4Token(httpHelper); - } - - Map remoteConfigStatus = new HashMap<>(); - - Map heads = new HashMap<>(); - heads.put("X-Token", token); - String response = httpHelper.get(baseUrl + "/api/v1/canal/instances", heads, REQUEST_TIMEOUT); - ResponseModel> resp = JSONObject.parseObject(response, - new TypeReference>>() { - }); - if (HttpHelper.REST_STATE_TOKEN_INVALID.equals(resp.getCode())) { - // token 失效 - token = login4Token(httpHelper); - heads.put("X-Token", token); - response = httpHelper.get(baseUrl + "/api/v1/canal/instances", heads, REQUEST_TIMEOUT); - resp = JSONObject.parseObject(response, new TypeReference>>() { - }); - } - if (!HttpHelper.REST_STATE_OK.equals(resp.getCode())) { - throw new RuntimeException("requestGet for canal instances error: " + resp.getMessage()); - } - for (ConfigItem configItem : resp.getData()) { - remoteConfigStatus.put(configItem.getName(), configItem); - } - - if (!remoteConfigStatus.isEmpty()) { - List changedIds = new ArrayList<>(); - - for (ConfigItem remoteConfigStat : remoteConfigStatus.values()) { - ConfigItem currentConfig = remoteInstanceConfigs.get(remoteConfigStat.getName()); - if (currentConfig == null) { - // 新增 - changedIds.add(remoteConfigStat.getId()); - } else { - // 修改 - if (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { - changedIds.add(remoteConfigStat.getId()); - } - } - } - if (!changedIds.isEmpty()) { - for (Long changedId : changedIds) { - String response2 = httpHelper - .get(baseUrl + "/api/v1/canal/instance/" + changedId, heads, REQUEST_TIMEOUT); - ResponseModel resp2 = JSONObject.parseObject(response2, - new TypeReference>() { - }); - ConfigItem configItemNew = resp2.getData(); - remoteInstanceConfigs.put(configItemNew.getName(), configItemNew); - remoteInstanceMonitor.onModify(configItemNew); - } - } - } - - for (String name : remoteInstanceConfigs.keySet()) { - if (!remoteConfigStatus.containsKey(name)) { - // 删除 - remoteInstanceConfigs.remove(name); - remoteInstanceMonitor.onDelete(name); - } - } - } - - @Override - public synchronized void startMonitor(RemoteCanalConfigMonitor remoteCanalConfigMonitor) { - - executor.scheduleWithFixedDelay(new Runnable() { - - public void run() { - // 监听canal.properties变化 - try { - Properties properties = loadRemoteConfig(); - if (properties != null) { - remoteCanalConfigMonitor.onChange(properties); - } - } catch (Throwable e) { - logger.error("Scan remote canal config failed", e); - } - - // 监听instance变化 - try { - loadRemoteInstanceConfigs(); - } catch (Throwable e) { - logger.error("Scan remote instance config failed", e); - } - } - - }, 10, scanIntervalInSecond, TimeUnit.SECONDS); - } - - /** - * 覆盖本地 canal.properties - * - * @param content 远程配置内容文本 - */ - private void overrideLocalCanalConfig(String content) { - try (FileWriter writer = new FileWriter(CommonUtils.getConfPath() + "canal.properties")) { - writer.write(content); - writer.flush(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - @Override - public synchronized void destroy() { - if (httpHelper != null) { - httpHelper.close(); - } - } -} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/ResponseModel.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/ResponseModel.java deleted file mode 100644 index 22c42be7..00000000 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/ResponseModel.java +++ /dev/null @@ -1,38 +0,0 @@ -package com.alibaba.otter.canal.deployer.monitor.remote.http; - -/** - * 响应类 - * - * @author rewerma 2019-08-26 上午09:40:36 - * @version 1.0.0 - */ -public class ResponseModel { - - private Integer code; - private String message; - private T data; - - public Integer getCode() { - return code; - } - - public void setCode(Integer code) { - this.code = code; - } - - public String getMessage() { - return message; - } - - public void setMessage(String message) { - this.message = message; - } - - public T getData() { - return data; - } - - public void setData(T data) { - this.data = data; - } -} diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 4a750e89..9d2fff8e 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -1,13 +1,6 @@ ################################################# ######### common argument ############# ################################################# -#canal.manager.jdbc.url=jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8 -#canal.manager.jdbc.username=root -#canal.manager.jdbc.password=121212 -canal.manager.http.url=http://127.0.0.1:8089 -canal.manager.http.username=admin -canal.manager.http.password=121212 - canal.id = 1 # tcp bind ip canal.ip = @@ -20,6 +13,7 @@ canal.metrics.pull.port = 11112 # canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458 # canal admin config +#canal.admin.manager = 127.0.0.1:8089 canal.admin.port = 11110 canal.admin.user = admin canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441 @@ -112,7 +106,7 @@ canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml canal.instance.global.mode = spring canal.instance.global.lazy = false -#canal.instance.global.manager.address = 127.0.0.1:1099 +canal.instance.global.manager.address = ${canal.admin.manager} #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml canal.instance.global.spring.xml = classpath:spring/file-instance.xml #canal.instance.global.spring.xml = classpath:spring/default-instance.xml diff --git a/instance/manager/pom.xml b/instance/manager/pom.xml index a8b8856b..678632d7 100644 --- a/instance/manager/pom.xml +++ b/instance/manager/pom.xml @@ -16,5 +16,16 @@ canal.instance.core ${project.version} + + com.alibaba.otter + canal.instance.spring + ${project.version} + + + + junit + junit + test + diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java new file mode 100644 index 00000000..ee3f99bd --- /dev/null +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java @@ -0,0 +1,72 @@ +package com.alibaba.otter.canal.instance.manager; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +import com.alibaba.otter.canal.common.CanalException; +import com.alibaba.otter.canal.instance.core.CanalInstance; +import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanal; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient; +import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator; + +/** + * 基于manager生成对应的{@linkplain CanalInstance} + * + * @author jianghang 2012-7-12 下午05:37:09 + * @version 1.0.0 + */ +public class PlainCanalInstanceGenerator implements CanalInstanceGenerator { + + private static final Logger logger = LoggerFactory.getLogger(SpringCanalInstanceGenerator.class); + private String springXml; + private PlainCanalConfigClient canalConfigClient; + private String defaultName = "instance"; + private BeanFactory beanFactory; + + public CanalInstance generate(String destination) { + synchronized (CanalInstanceGenerator.class) { + try { + PlainCanal canal = canalConfigClient.findInstance(destination, null); + if (canal == null) { + throw new CanalException("instance : " + destination + " config is not found"); + } + // 设置动态properties,替换掉本地properties + com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer.propertiesLocal.set(canal.getProperties()); + // 设置当前正在加载的通道，加载spring查找文件时会用到该变量 + System.setProperty("canal.instance.destination", destination); + this.beanFactory = getBeanFactory(springXml); + String beanName = destination; + if (!beanFactory.containsBean(beanName)) { + beanName = defaultName; + } + + return (CanalInstance) beanFactory.getBean(beanName); + } catch (Throwable e) { + logger.error("generator instance failed.", e); + throw new CanalException(e); + } finally { + System.setProperty("canal.instance.destination", ""); + } + } + } + + // ================ setter / getter ================ + + private BeanFactory getBeanFactory(String springXml) { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml); + return applicationContext; + } + + public void setCanalConfigClient(PlainCanalConfigClient canalConfigClient) { + this.canalConfigClient = canalConfigClient; + } + + public void setSpringXml(String springXml) { + this.springXml = springXml; + } + +} diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpHelper.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/HttpHelper.java similarity index 77% rename from deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpHelper.java rename to instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/HttpHelper.java index 516241af..af8dcf34 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/remote/http/HttpHelper.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/HttpHelper.java @@ -1,9 +1,15 @@ -package com.alibaba.otter.canal.deployer.monitor.remote.http; +package com.alibaba.otter.canal.instance.manager.plain; + +import static org.apache.http.client.config.RequestConfig.custom; + +import java.io.IOException; +import java.net.URI; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; +import java.util.Map; + +import javax.net.ssl.SSLContext; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONObject; -import com.alibaba.fastjson.TypeReference; -import com.alibaba.otter.canal.deployer.monitor.remote.ConfigItem; import org.apache.http.HttpStatus; import org.apache.http.client.config.RequestConfig; import org.apache.http.client.methods.CloseableHttpResponse; @@ -11,18 +17,22 @@ import org.apache.http.client.methods.HttpGet; import org.apache.http.client.methods.HttpPost; import org.apache.http.client.protocol.HttpClientContext; import org.apache.http.client.utils.URIBuilder; +import org.apache.http.config.RegistryBuilder; +import org.apache.http.conn.socket.ConnectionSocketFactory; +import org.apache.http.conn.socket.PlainConnectionSocketFactory; +import org.apache.http.conn.ssl.NoopHostnameVerifier; +import org.apache.http.conn.ssl.SSLConnectionSocketFactory; import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; +import org.apache.http.ssl.SSLContextBuilder; +import org.apache.http.ssl.TrustStrategy; import org.apache.http.util.EntityUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.net.URI; -import java.util.Map; - -import static org.apache.http.client.config.RequestConfig.custom; +import com.alibaba.fastjson.JSON; /** * http client 工具类 @@ -32,7 +42,7 @@ import static org.apache.http.client.config.RequestConfig.custom; */ public class HttpHelper { - private final static Logger logger = LoggerFactory.getLogger(HttpRemoteConfigLoader.class); + private final static Logger logger = LoggerFactory.getLogger(HttpHelper.class); public static final Integer REST_STATE_OK = 20000; public static final Integer REST_STATE_TOKEN_INVALID = 50014; @@ -44,16 +54,31 @@ public class HttpHelper { HttpClientBuilder builder = HttpClientBuilder.create(); builder.setMaxConnPerRoute(50); builder.setMaxConnTotal(100); - httpclient = builder.build(); + + // 创建支持忽略证书的https + try { + SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() { + + @Override + public boolean isTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException { + return true; + } + }).build(); + + httpclient = HttpClientBuilder.create() + .setSSLContext(sslContext) + .setConnectionManager(new PoolingHttpClientConnectionManager(RegistryBuilder. create() + .register("http", PlainConnectionSocketFactory.INSTANCE) + .register("https", new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE)) + .build())) + .build(); + } catch (Throwable e) { + // ignore + } } public String get(String url, Map heads, int timeout) { - // 支持采用https协议，忽略证书 url = url.trim(); - if (url.startsWith("https")) { - // FIXME - return ""; - } CloseableHttpResponse response = null; HttpGet httpGet = null; try { @@ -97,17 +122,11 @@ public class HttpHelper { } public String post(String url, Map heads, Object requestBody, int timeout) { - String json = JSON.toJSONString(requestBody); - return post0(url, heads, json, timeout); + return post0(url, heads, JSON.toJSONString(requestBody), timeout); } public String post0(String url, Map heads, String requestBody, int timeout) { url = url.trim(); - // 支持采用https协议，忽略证书 - if (url.startsWith("https")) { - // FIXME - return ""; - } HttpPost httpPost = null; CloseableHttpResponse response = null; try { diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanal.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanal.java new file mode 100644 index 00000000..871c7ed5 --- /dev/null +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanal.java @@ -0,0 +1,55 @@ +package com.alibaba.otter.canal.instance.manager.plain; + +import java.util.Properties; + +/** + * plain远程配置，提供基于properties纯文本的配置 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @author agapple + * @version 1.0.0 + */ +public class PlainCanal { + + private Properties properties; + private String md5; + private String status; + + public PlainCanal(){ + } + + public PlainCanal(Properties properties, String status, String md5){ + this.properties = properties; + this.md5 = md5; + this.status = status; + } + + public Properties getProperties() { + return properties; + } + + public void setProperties(Properties properties) { + this.properties = properties; + } + + public String getMd5() { + return md5; + } + + public void setMd5(String md5) { + this.md5 = md5; + } + + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + + @Override + public String toString() { + return "PlainCanal [properties=" + properties + ", md5=" + md5 + ", status=" + status + "]"; + } +} diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java new file mode 100644 index 00000000..bdf2264b --- /dev/null +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java @@ -0,0 +1,141 @@ +package com.alibaba.otter.canal.instance.manager.plain; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.NoSuchAlgorithmException; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +import org.apache.commons.lang.StringUtils; + +import com.alibaba.fastjson.JSONObject; +import com.alibaba.fastjson.TypeReference; +import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; +import com.alibaba.otter.canal.common.CanalException; +import com.alibaba.otter.canal.common.CanalLifeCycle; +import com.alibaba.otter.canal.protocol.SecurityUtil; + +/** + * 远程配置获取 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @author agapple 2019年8月26日 下午7:52:06 + * @since 1.1.4 + */ +public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements CanalLifeCycle { + + private final static Integer REQUEST_TIMEOUT = 5000; + private String configURL; + private String user; + private String passwd; + private HttpHelper httpHelper; + + public PlainCanalConfigClient(String configURL, String user, String passwd){ + this.configURL = configURL; + if (!StringUtils.startsWithIgnoreCase(configURL, "http")) { + this.configURL = "http://" + configURL; + } else { + this.configURL = configURL; + } + this.user = user; + this.passwd = passwd; + this.httpHelper = new HttpHelper(); + } + + /** + * 加载canal.properties文件 + * + * @return 远程配置的properties + */ + public PlainCanal findServer(String md5) { + if (StringUtils.isEmpty(md5)) { + md5 = ""; + } + String url = configURL + "/api/v1/config/server_poll?md5=" + md5; + return queryConfig(url); + } + + /** + * 加载远程的instance.properties + */ + public PlainCanal findInstance(String destination, String md5) { + if (StringUtils.isEmpty(md5)) { + md5 = ""; + } + String url = configURL + "/api/v1/config/instance_poll/" + destination + "?md5=" + md5; + return queryConfig(url); + } + + /** + * 返回需要运行的instance列表 + */ + public String findInstances(String ip, String port, String md5) { + if (StringUtils.isEmpty(md5)) { + md5 = ""; + } + String url = configURL + "/api/v1/config/instances_poll?md5=" + md5 + "&ip=" + ip + "&port=" + port; + ResponseModel config = doQuery(url); + if (config.data != null) { + return config.data.content; + } else { + return null; + } + } + + private PlainCanal queryConfig(String url) { + try { + ResponseModel config = doQuery(url); + return processData(config.data); + } catch (Throwable e) { + throw new CanalException("load manager config failed.", e); + } + } + + private ResponseModel doQuery(String url) { + Map heads = new HashMap<>(); + heads.put("user", user); + heads.put("passwd", passwd); + String response = httpHelper.get(url, heads, REQUEST_TIMEOUT); + ResponseModel resp = JSONObject.parseObject(response, + new TypeReference>() { + }); + + if (!HttpHelper.REST_STATE_OK.equals(resp.code)) { + throw new CanalException("requestGet for canal config error: " + resp.message); + } + + return resp; + } + + private PlainCanal processData(CanalConfig config) throws IOException, NoSuchAlgorithmException { + Properties properties = new Properties(); + String md5 = null; + String status = null; + if (config != null && StringUtils.isNotEmpty(config.content)) { + md5 = SecurityUtil.md5String(config.content); + status = config.status; + properties.load(new ByteArrayInputStream(config.content.getBytes(StandardCharsets.UTF_8))); + } else { + // null代表没有新配置变更 + return null; + } + + return new PlainCanal(properties, status, md5); + } + + private static class ResponseModel { + + public Integer code; + public String message; + public T data; + } + + private static class CanalConfig { + + public String content; + public String status; + + } +} diff --git a/instance/manager/src/test/java/com/alibaba/otter/canal/instance/manager/PlainCanalConfigClientIntegration.java b/instance/manager/src/test/java/com/alibaba/otter/canal/instance/manager/PlainCanalConfigClientIntegration.java new file mode 100644 index 00000000..1ed3651c --- /dev/null +++ b/instance/manager/src/test/java/com/alibaba/otter/canal/instance/manager/PlainCanalConfigClientIntegration.java @@ -0,0 +1,29 @@ +package com.alibaba.otter.canal.instance.manager; + +import org.junit.Test; +import org.springframework.util.Assert; + +import com.alibaba.otter.canal.instance.manager.plain.PlainCanal; +import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient; + +public class PlainCanalConfigClientIntegration { + + @Test + public void testSimple() { + PlainCanalConfigClient client = new PlainCanalConfigClient("http://127.0.0.1:8089", + "admin", + "4ACFE3202A5FF5CF467898FC58AAB1D615029441"); + + PlainCanal plain = client.findServer(null); + Assert.notNull(plain); + + plain = client.findServer(plain.getMd5()); + Assert.isNull(plain); + + plain = client.findInstance("example", null); + Assert.notNull(plain); + + plain = client.findInstance("example", plain.getMd5()); + Assert.isNull(plain); + } +} diff --git a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/SpringCanalInstanceGenerator.java b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/SpringCanalInstanceGenerator.java index 690c084a..a5e7f034 100644 --- a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/SpringCanalInstanceGenerator.java +++ b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/SpringCanalInstanceGenerator.java @@ -1,9 +1,12 @@ package com.alibaba.otter.canal.instance.spring; -import org.springframework.beans.BeansException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import com.alibaba.otter.canal.common.CanalException; import com.alibaba.otter.canal.instance.core.CanalInstance; import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator; @@ -11,22 +14,40 @@ import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator; * @author zebin.xuzb @ 2012-7-12 * @version 1.0.0 */ -public class SpringCanalInstanceGenerator implements CanalInstanceGenerator, BeanFactoryAware { +public class SpringCanalInstanceGenerator implements CanalInstanceGenerator { - private String defaultName = "instance"; - private BeanFactory beanFactory; + private static final Logger logger = LoggerFactory.getLogger(SpringCanalInstanceGenerator.class); + private String springXml; + private String defaultName = "instance"; + private BeanFactory beanFactory; public CanalInstance generate(String destination) { - String beanName = destination; - if (!beanFactory.containsBean(beanName)) { - beanName = defaultName; + synchronized (CanalInstanceGenerator.class) { + try { + // 设置当前正在加载的通道，加载spring查找文件时会用到该变量 + System.setProperty("canal.instance.destination", destination); + this.beanFactory = getBeanFactory(springXml); + String beanName = destination; + if (!beanFactory.containsBean(beanName)) { + beanName = defaultName; + } + + return (CanalInstance) beanFactory.getBean(beanName); + } catch (Throwable e) { + logger.error("generator instance failed.", e); + throw new CanalException(e); + } finally { + System.setProperty("canal.instance.destination", ""); + } } - - return (CanalInstance) beanFactory.getBean(beanName); } - public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - this.beanFactory = beanFactory; + private BeanFactory getBeanFactory(String springXml) { + ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml); + return applicationContext; } + public void setSpringXml(String springXml) { + this.springXml = springXml; + } } diff --git a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/support/PropertyPlaceholderConfigurer.java b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/support/PropertyPlaceholderConfigurer.java index 0c3a5518..f4b7be5b 100644 --- a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/support/PropertyPlaceholderConfigurer.java +++ b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/support/PropertyPlaceholderConfigurer.java @@ -21,10 +21,19 @@ import org.springframework.util.Assert; */ public class PropertyPlaceholderConfigurer extends org.springframework.beans.factory.config.PropertyPlaceholderConfigurer implements ResourceLoaderAware, InitializingBean { - private static final String PLACEHOLDER_PREFIX = "${"; - private static final String PLACEHOLDER_SUFFIX = "}"; - private ResourceLoader loader; - private String[] locationNames; + private static final String PLACEHOLDER_PREFIX = "${"; + private static final String PLACEHOLDER_SUFFIX = "}"; + public static ThreadLocal propertiesLocal = new ThreadLocal() { + + @Override + protected Properties initialValue() { + return new Properties(); + } + + }; + + private ResourceLoader loader; + private String[] locationNames; public PropertyPlaceholderConfigurer(){ setIgnoreUnresolvablePlaceholders(true); @@ -117,13 +126,31 @@ public class PropertyPlaceholderConfigurer extends org.springframework.beans.fac @Override protected String resolvePlaceholder(String placeholder, Properties props, int systemPropertiesMode) { DefaultablePlaceholder dp = new DefaultablePlaceholder(placeholder); - String value = super.resolvePlaceholder(dp.placeholder, props, systemPropertiesMode); - - if (value == null) { - value = dp.defaultValue; + String propVal = null; + // 以system为准覆盖本地配置, 适用于docker + if (systemPropertiesMode == SYSTEM_PROPERTIES_MODE_OVERRIDE) { + propVal = resolveSystemProperty(dp.placeholder); } - return trimToEmpty(value); + // 以threadlocal的为准覆盖file properties + if (propVal == null) { + Properties localProperties = propertiesLocal.get(); + propVal = resolvePlaceholder(dp.placeholder, localProperties); + } + + if (propVal == null) { + propVal = resolvePlaceholder(dp.placeholder, props); + } + + if (propVal == null && systemPropertiesMode == SYSTEM_PROPERTIES_MODE_FALLBACK) { + propVal = resolveSystemProperty(dp.placeholder); + } + + if (propVal == null) { + propVal = dp.defaultValue; + } + + return trimToEmpty(propVal); } private static class DefaultablePlaceholder { diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java index eec3ad53..18686774 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java @@ -3,6 +3,8 @@ package com.alibaba.otter.canal.protocol; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; /** * 
@@ -19,6 +21,25 @@ import java.util.Arrays;
  */
 public class SecurityUtil {
 
+    private static char[]                  digits  = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c',
+            'd', 'e', 'f'                         };
+
+    private static Map rDigits = new HashMap(16);
+    static {
+        for (int i = 0; i < digits.length; ++i) {
+            rDigits.put(digits[i], i);
+        }
+    }
+
+    public static String md5String(String content) throws NoSuchAlgorithmException {
+        MessageDigest md = MessageDigest.getInstance("md5");
+        byte[] bt = md.digest(content.getBytes());
+        if (null == bt || bt.length != 16) {
+            throw new IllegalArgumentException("md5 need");
+        }
+        return byte2HexStr(bt);
+    }
+
     public static final String scrambleGenPass(byte[] pass) throws NoSuchAlgorithmException {
         MessageDigest md = MessageDigest.getInstance("SHA-1");
         byte[] pass1 = md.digest(pass);