mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
自动注册时可自定义node server名称,canal.properties#canal.admin.register.name,为空则使用canal.register.ip (#3459)
This commit is contained in:
@@ -46,14 +46,15 @@ public class PollingConfigController {
|
||||
public BaseModel<CanalConfig> canalConfigPoll(@RequestHeader String user, @RequestHeader String passwd,
|
||||
@RequestParam String ip, @RequestParam Integer port,
|
||||
@RequestParam String md5, @RequestParam boolean register,
|
||||
@RequestParam String cluster, @PathVariable String env) {
|
||||
@RequestParam String cluster, @RequestParam String name,
|
||||
@PathVariable String env) {
|
||||
if (!auth(user, passwd)) {
|
||||
throw new RuntimeException("auth :" + user + " is failed");
|
||||
}
|
||||
|
||||
if (StringUtils.isEmpty(md5) && register) {
|
||||
// do something
|
||||
pollingConfigService.autoRegister(ip, port, cluster);
|
||||
pollingConfigService.autoRegister(ip, port, cluster, StringUtils.trimToNull(name));
|
||||
}
|
||||
|
||||
CanalConfig canalConfig = pollingConfigService.getChangedConfig(ip, port, md5);
|
||||
|
||||
@@ -5,7 +5,7 @@ import com.alibaba.otter.canal.admin.model.CanalInstanceConfig;
|
||||
|
||||
public interface PollingConfigService {
|
||||
|
||||
public boolean autoRegister(String ip, Integer adminPort, String cluster);
|
||||
public boolean autoRegister(String ip, Integer adminPort, String cluster, String name);
|
||||
|
||||
CanalConfig getChangedConfig(String ip, Integer port, String md5);
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.admin.service.impl;
|
||||
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
@@ -28,11 +29,11 @@ public class PollingConfigServiceImpl implements PollingConfigService {
|
||||
@Autowired
|
||||
CanalClusterService canalClusterService;
|
||||
|
||||
public boolean autoRegister(String ip, Integer adminPort, String cluster) {
|
||||
public boolean autoRegister(String ip, Integer adminPort, String cluster, String name) {
|
||||
NodeServer server = NodeServer.find.query().where().eq("ip", ip).eq("adminPort", adminPort).findOne();
|
||||
if (server == null) {
|
||||
server = new NodeServer();
|
||||
server.setName(ip);
|
||||
server.setName(Optional.ofNullable(name).orElse(ip));
|
||||
server.setIp(ip);
|
||||
server.setAdminPort(adminPort);
|
||||
server.setTcpPort(adminPort + 1);
|
||||
|
||||
@@ -16,6 +16,10 @@ canal.metrics.pull.port = 11112
|
||||
canal.admin.port = 11110
|
||||
canal.admin.user = admin
|
||||
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
|
||||
# admin auto register
|
||||
#canal.admin.register.auto = true
|
||||
#canal.admin.register.cluster =
|
||||
#canal.admin.register.name =
|
||||
|
||||
canal.zkServers =
|
||||
# flush data to zk
|
||||
|
||||
@@ -25,6 +25,7 @@ public class CanalConstants {
|
||||
public static final String CANAL_ADMIN_PASSWD = ROOT + "." + "admin.passwd";
|
||||
public static final String CANAL_ADMIN_AUTO_REGISTER = ROOT + "." + "admin.register.auto";
|
||||
public static final String CANAL_ADMIN_AUTO_CLUSTER = ROOT + "." + "admin.register.cluster";
|
||||
public static final String CANAL_ADMIN_REGISTER_NAME = ROOT + "." + "admin.register.name";
|
||||
public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers";
|
||||
public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty";
|
||||
|
||||
|
||||
@@ -55,6 +55,7 @@ public class CanalLauncher {
|
||||
boolean autoRegister = BooleanUtils.toBoolean(CanalController.getProperty(properties,
|
||||
CanalConstants.CANAL_ADMIN_AUTO_REGISTER));
|
||||
String autoCluster = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_AUTO_CLUSTER);
|
||||
String name = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_REGISTER_NAME);
|
||||
String registerIp = CanalController.getProperty(properties, CanalConstants.CANAL_REGISTER_IP);
|
||||
if (StringUtils.isEmpty(registerIp)) {
|
||||
registerIp = AddressUtils.getHostIp();
|
||||
@@ -65,7 +66,8 @@ public class CanalLauncher {
|
||||
registerIp,
|
||||
Integer.parseInt(adminPort),
|
||||
autoRegister,
|
||||
autoCluster);
|
||||
autoCluster,
|
||||
name);
|
||||
PlainCanal canalConfig = configClient.findServer(null);
|
||||
if (canalConfig == null) {
|
||||
throw new IllegalArgumentException("managerAddress:" + managerAddress
|
||||
|
||||
@@ -8,4 +8,5 @@ canal.admin.user = admin
|
||||
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
|
||||
# admin auto register
|
||||
canal.admin.register.auto = true
|
||||
canal.admin.register.cluster =
|
||||
canal.admin.register.cluster =
|
||||
canal.admin.register.name =
|
||||
@@ -35,12 +35,14 @@ public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements Ca
|
||||
private int adminPort;
|
||||
private boolean autoRegister;
|
||||
private String autoCluster;
|
||||
private String name;
|
||||
|
||||
public PlainCanalConfigClient(String configURL, String user, String passwd, String localIp, int adminPort,
|
||||
boolean autoRegister, String autoCluster){
|
||||
boolean autoRegister, String autoCluster, String name){
|
||||
this(configURL, user, passwd, localIp, adminPort);
|
||||
this.autoCluster = autoCluster;
|
||||
this.autoRegister = autoRegister;
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public PlainCanalConfigClient(String configURL, String user, String passwd, String localIp, int adminPort){
|
||||
@@ -71,7 +73,7 @@ public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements Ca
|
||||
md5 = "";
|
||||
}
|
||||
String url = configURL + "/api/v1/config/server_polling?ip=" + localIp + "&port=" + adminPort + "&md5=" + md5
|
||||
+ "®ister=" + (autoRegister ? 1 : 0) + "&cluster=" + autoCluster;
|
||||
+ "®ister=" + (autoRegister ? 1 : 0) + "&cluster=" + autoCluster + "&name=" + name;
|
||||
return queryConfig(url);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user