chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed issue #2099 , support user/passwd acl and canal admin operator

Browse Source
This commit is contained in:
agapple
2019-08-26 18:16:33 +08:00
parent e376a129ed
commit 3d4dbc9a8c
54 changed files with 8518 additions and 813 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
View File

@@ -16,8 +16,12 @@ public class CanalConstants {
public static final String CANAL_IP = ROOT + "." + "ip";
public static final String CANAL_REGISTER_IP = ROOT + "." + "register.ip";
public static final String CANAL_PORT = ROOT + "." + "port";
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_JMX_PORT = ROOT + "." + "admin.jmx.port";
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";
public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers";
public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty";

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

@@ -121,6 +121,9 @@ public class CanalController {
embededCanalServer.setMetricsPort(11112);
}
embededCanalServer.setUser(getProperty(properties, CanalConstants.CANAL_USER));
embededCanalServer.setPasswd(getProperty(properties, CanalConstants.CANAL_PASSWD));
String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY);
if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) {
canalServer = CanalServerWithNetty.instance();
@@ -571,4 +574,5 @@ public class CanalController {
public Map<String, InstanceConfig> getInstanceConfigs() {
return instanceConfigs;
}
}

23
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
View File

@@ -8,9 +8,6 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.deployer.mbean.CanalServerAgent;
import com.alibaba.otter.canal.deployer.mbean.CanalServerBean;
import com.alibaba.otter.canal.deployer.mbean.CanalServerMXBean;
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;
@@ -76,27 +73,7 @@ public class CanalLauncher {
});
}
CanalServerAgent canalServerAgent = null;
String jmxPort = properties.getProperty(CanalConstants.CANAL_ADMIN_JMX_PORT);
if (StringUtils.isNotEmpty(jmxPort)) {
String ip = properties.getProperty(CanalConstants.CANAL_IP);
String registerIp = properties.getProperty(CanalConstants.CANAL_REGISTER_IP);
if (StringUtils.isEmpty(registerIp)) {
// 兼容老的配置
registerIp = ip;
}
CanalServerMXBean canalServerMBean = new CanalServerBean(canalStater);
canalServerAgent = new CanalServerAgent(registerIp, Integer.parseInt(jmxPort), canalServerMBean);
Thread agentThread = new Thread(canalServerAgent::start);
agentThread.start();
}
runningLatch.await();
if (canalServerAgent != null) {
canalServerAgent.stop();
}
if (remoteConfigLoader != null) {
remoteConfigLoader.destroy();
}

33
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java
View File

@@ -10,7 +10,9 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.admin.netty.CanalAdminWithNetty;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.deployer.admin.CanalAdminController;
import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer;
import com.alibaba.otter.canal.server.CanalMQStarter;
@@ -36,6 +38,8 @@ public class CanalStater {
private volatile Properties properties;
private volatile boolean running = false;
private CanalAdminWithNetty canalAdmin;
public CanalStater(Properties properties){
this.properties = properties;
}
@@ -136,15 +140,42 @@ public class CanalStater {
controller.setCanalMQStarter(canalMQStarter);
}
// start canalAdmin
String port = properties.getProperty(CanalConstants.CANAL_ADMIN_PORT);
if (canalAdmin == null && StringUtils.isNotEmpty(port)) {
String user = properties.getProperty(CanalConstants.CANAL_ADMIN_USER);
String passwd = properties.getProperty(CanalConstants.CANAL_ADMIN_PASSWD);
CanalAdminController canalAdmin = new CanalAdminController(this);
canalAdmin.setUser(user);
canalAdmin.setPasswd(passwd);
String ip = properties.getProperty(CanalConstants.CANAL_IP);
CanalAdminWithNetty canalAdminWithNetty = CanalAdminWithNetty.instance();
canalAdminWithNetty.setCanalAdmin(canalAdmin);
canalAdminWithNetty.setPort(Integer.valueOf(port));
canalAdminWithNetty.setIp(ip);
canalAdminWithNetty.start();
this.canalAdmin = canalAdminWithNetty;
}
running = true;
}
public synchronized void stop() throws Throwable {
stop(false);
}
/**
* 销毁方法，远程配置变更时调用
*
* @throws Throwable
*/
public synchronized void stop() throws Throwable {
public synchronized void stop(boolean stopByAdmin) throws Throwable {
if (!stopByAdmin && canalAdmin != null) {
canalAdmin.stop();
canalAdmin = null;
}
if (controller != null) {
controller.stop();
controller = null;

224
deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java Normal file
View File

@@ -0,0 +1,224 @@
package com.alibaba.otter.canal.deployer.admin;
import java.io.File;
import java.security.NoSuchAlgorithmException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.admin.CanalAdmin;
import com.alibaba.otter.canal.common.utils.FileUtils;
import com.alibaba.otter.canal.deployer.CanalController;
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.SpringInstanceConfigMonitor;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.protocol.SecurityUtil;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.google.common.base.Joiner;
/**
* 提供canal admin的管理操作
*
* @author agapple 2019年8月24日 下午11:39:01
* @since 1.1.4
*/
public class CanalAdminController implements CanalAdmin {
private static final Logger logger = LoggerFactory.getLogger(CanalAdminController.class);
private String user;
private String passwd;
private CanalStater canalStater;
public CanalAdminController(CanalStater canalStater){
this.canalStater = canalStater;
}
@Override
public boolean check() {
return canalStater.isRunning();
}
@Override
public synchronized boolean start() {
try {
if (!canalStater.isRunning()) {
canalStater.start();
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean stop() {
try {
if (canalStater.isRunning()) {
canalStater.stop(true);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean restart() {
stop();
return start();
}
@Override
public boolean auth(String user, String passwd, byte[] seed) {
// 如果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 {
byte[] passForClient = SecurityUtil.hexStr2Bytes(passwd);
return SecurityUtil.scrambleServerAuth(passForClient, SecurityUtil.hexStr2Bytes(this.passwd), seed);
} catch (NoSuchAlgorithmException e) {
return false;
}
}
return false;
}
@Override
public String getRunningInstances() {
try {
CanalController controller = canalStater.getController();
if (controller != null) {
Map<String, InstanceConfig> instanceConfigs = controller.getInstanceConfigs();
if (instanceConfigs != null) {
return Joiner.on(",").join(instanceConfigs.keySet());
}
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return "";
}
@Override
public boolean checkInstance(String destination) {
Map<String, CanalInstance> instances = CanalServerWithEmbedded.instance().getCanalInstances();
if (instances == null || !instances.containsKey(destination)) {
return false;
} else {
CanalInstance instance = instances.get(destination);
return instance.isStart();
}
}
@Override
public boolean startInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.start(destination);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public boolean stopInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.stop(destination);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public boolean restartInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.reload(destination);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public String listCanalLog() {
Collection<File> files = org.apache.commons.io.FileUtils.listFiles(new File("../logs/canal/"),
TrueFileFilter.TRUE,
TrueFileFilter.TRUE);
List<String> names = files.stream().map(f -> f.getName()).collect(Collectors.toList());
return Joiner.on(",").join(names);
}
@Override
public String canalLog(int lines) {
return FileUtils.readFileFromOffset("../logs/canal/canal.log", lines, "UTF-8");
}
@Override
public String listInstanceLog(String destination) {
Collection<File> files = org.apache.commons.io.FileUtils.listFiles(new File("../logs/" + destination + "/"),
TrueFileFilter.TRUE,
TrueFileFilter.TRUE);
List<String> names = files.stream().map(f -> f.getName()).collect(Collectors.toList());
return Joiner.on(",").join(names);
}
@Override
public String instanceLog(String destination, String fileName, int lines) {
if (StringUtils.isEmpty(fileName)) {
fileName = destination + ".log";
}
return FileUtils.readFileFromOffset("../logs/" + destination + "/" + fileName, lines, "UTF-8");
}
private InstanceAction getInstanceAction(String destination) {
Map<InstanceConfig.InstanceMode, InstanceConfigMonitor> monitors = canalStater.getController()
.getInstanceConfigMonitors();
SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors.get(InstanceConfig.InstanceMode.SPRING);
Map<String, InstanceAction> instanceActions = monitor.getActions();
return instanceActions.get(destination);
}
public void setUser(String user) {
this.user = user;
}
public void setPasswd(String passwd) {
this.passwd = passwd;
}
public void setCanalStater(CanalStater canalStater) {
this.canalStater = canalStater;
}
}

79
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerAgent.java
View File

@@ -1,79 +0,0 @@
package com.alibaba.otter.canal.deployer.mbean;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.rmi.registry.LocateRegistry;
import javax.management.MBeanServer;
import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.ObjectName;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;
/**
* Canal Server Agent 用于远程JMX调用
*
* @author rewerma 2019-07-13 下午05:12:16
* @version 1.0.0
*/
public class CanalServerAgent implements NotificationListener {
private static final Logger logger = LoggerFactory.getLogger(CanalServerBean.class);
private MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
private String ip = "0.0.0.0";
private int port;
private CanalServerMXBean canalServerMbean;
private volatile JMXConnectorServer cs;
public CanalServerAgent(String ip, int port, CanalServerMXBean canalServerMbean){
if (StringUtils.isNotEmpty(ip)) {
this.ip = ip;
}
this.port = port;
this.canalServerMbean = canalServerMbean;
}
@Override
public void handleNotification(Notification notification, Object handback) {
}
public synchronized void start() {
try {
if (cs == null) {
ObjectName name = new ObjectName("CanalServerAgent:type=CanalServerStatus");
mBeanServer.registerMBean(canalServerMbean, name);
LocateRegistry.createRegistry(port);
JMXServiceURL jmxServiceURL = new JMXServiceURL(
"service:jmx:rmi:///jndi/rmi://" + ip + ":" + port + "/jmxrmi");
cs = JMXConnectorServerFactory.newJMXConnectorServer(jmxServiceURL, null, mBeanServer);
cs.start();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
public synchronized void stop() {
if (cs != null) {
try {
cs.stop();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}

156
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java
View File

@@ -1,156 +0,0 @@
package com.alibaba.otter.canal.deployer.mbean;
import com.alibaba.otter.canal.common.utils.FileUtils;
import com.alibaba.otter.canal.deployer.CanalLauncher;
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.SpringInstanceConfigMonitor;
import com.google.common.base.Joiner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.management.NotificationBroadcasterSupport;
import java.util.Map;
/**
* Canal配置信息业务层
*
* @author rewerma 2019-07-13 下午05:12:16
* @version 1.0.0
*/
public class CanalServerBean extends NotificationBroadcasterSupport implements CanalServerMXBean {
private static final Logger logger = LoggerFactory.getLogger(CanalServerBean.class);
private volatile int status;
private CanalStater canalStater;
public CanalServerBean(CanalStater canalStater){
this.canalStater = canalStater;
this.status = canalStater.isRunning() ? 1 : 0;
}
@Override
public int getStatus() {
return status;
}
@Override
public synchronized boolean start() {
try {
if (!canalStater.isRunning()) {
canalStater.start();
status = 1;
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean stop() {
try {
if (canalStater.isRunning()) {
canalStater.stop();
status = 0;
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean restart() {
stop();
return start();
}
@Override
public synchronized boolean exit() {
stop();
CanalLauncher.runningLatch.countDown();
return true;
}
@Override
public synchronized boolean startInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.start(destination);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean stopInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.stop(destination);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean reloadInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.reload(destination);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized String getRunningInstances() {
try {
Map<String, InstanceConfig> instanceConfigs = canalStater.getController().getInstanceConfigs();
if (instanceConfigs != null) {
return Joiner.on(",").join(instanceConfigs.keySet());
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return "";
}
@Override
public String canalLog() {
return FileUtils.readFileFromOffset("../logs/canal/canal.log", 100, "UTF-8");
}
@Override
public String instanceLog(String destination) {
return FileUtils.readFileFromOffset("../logs/" + destination + "/" + destination + ".log", 100, "UTF-8");
}
private InstanceAction getInstanceAction(String destination) {
Map<InstanceConfig.InstanceMode, InstanceConfigMonitor> monitors = canalStater.getController()
.getInstanceConfigMonitors();
SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors
.get(InstanceConfig.InstanceMode.SPRING);
Map<String, InstanceAction> instanceActions = monitor.getActions();
return instanceActions.get(destination);
}
}

90
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerMXBean.java
View File

@@ -1,90 +0,0 @@
package com.alibaba.otter.canal.deployer.mbean;
/**
* Canal Server JMX MBean
*
* @author rewerma 2019-07-13 下午05:12:16
* @version 1.0.0
*/
public interface CanalServerMXBean {
/**
* 获取Canal Server状态
*
* @return 状态代码
*/
int getStatus();
/**
* 启动Canal Server
*
* @return 是否成功
*/
boolean start();
/**
* 停止Canal Server
*
* @return 是否成功
*/
boolean stop();
/**
* 重启Canal Server
*
* @return 是否成功
*/
boolean restart();
/**
* 退出Canal Server(关闭进程)
*
* @return 是否成功
*/
boolean exit();
/**
* 通过实例名启动实例
*
* @param destination 实例名
* @return 是否成功
*/
boolean startInstance(String destination);
/**
* 通过实例名关闭实例
*
* @param destination 实例名
* @return 是否成功
*/
boolean stopInstance(String destination);
/**
* 通过实例名重启实例
*
* @param destination 实例名
* @return 是否成功
*/
boolean reloadInstance(String destination);
/**
* 获取所有当前节点下运行中的实例
*
* @return 实例信息
*/
String getRunningInstances();
/**
* 获取Canal Server日志(末尾100行)
*
* @return 日志信息
*/
String canalLog();
/**
* 通过实例名获取实例日志(末尾100行)
*
* @return 日志信息
*/
String instanceLog(String destination);
}

12
deployer/src/main/resources/canal.properties
View File

@@ -11,7 +11,15 @@ canal.ip =
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.admin.jmx.port = 11113
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
# canal admin config
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
@@ -131,4 +139,4 @@ canal.mq.accessChannel = local
##################################################
canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"