Skip to content

Commit 862d017

Browse files
lulu2panpanlulu2panpan
committed
support auto generate instance names alibaba#4657 
1 parent fda52bc commit 862d017

File tree

4 files changed

+93
-29
lines changed

4 files changed

+93
-29
lines changed

‎deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java

+1
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ public class CanalConstants {
3030
public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty";
3131

3232
public static final String CANAL_DESTINATIONS = ROOT + "." + "destinations";
33+
public static final String CANAL_DESTINATIONS_EXPR = ROOT + "." + "destinations.expr";
3334
public static final String CANAL_AUTO_SCAN = ROOT + "." + "auto.scan";
3435
public static final String CANAL_AUTO_SCAN_INTERVAL = ROOT + "." + "auto.scan.interval";
3536
public static final String CANAL_CONF_DIR = ROOT + "." + "conf.dir";

‎deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java

+55-14
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,5 @@
11
package com.alibaba.otter.canal.deployer;
22

3-
import java.util.Map;
4-
import java.util.Properties;
5-
6-
import org.I0Itec.zkclient.IZkStateListener;
7-
import org.I0Itec.zkclient.exception.ZkNoNodeException;
8-
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
9-
import org.apache.commons.lang.BooleanUtils;
10-
import org.apache.commons.lang.StringUtils;
11-
import org.apache.zookeeper.Watcher.Event.KeeperState;
12-
import org.slf4j.Logger;
13-
import org.slf4j.LoggerFactory;
14-
import org.slf4j.MDC;
15-
163
import com.alibaba.otter.canal.common.utils.AddressUtils;
174
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
185
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
@@ -36,6 +23,26 @@
3623
import com.google.common.base.Function;
3724
import com.google.common.collect.MapMaker;
3825
import com.google.common.collect.MigrateMap;
26+
import org.I0Itec.zkclient.IZkStateListener;
27+
import org.I0Itec.zkclient.exception.ZkNoNodeException;
28+
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
29+
import org.apache.commons.lang.BooleanUtils;
30+
import org.apache.commons.lang.StringUtils;
31+
import org.apache.zookeeper.Watcher.Event.KeeperState;
32+
import org.slf4j.Logger;
33+
import org.slf4j.LoggerFactory;
34+
import org.slf4j.MDC;
35+
36+
import java.util.ArrayList;
37+
import java.util.List;
38+
import java.util.Map;
39+
import java.util.Properties;
40+
import java.util.regex.Matcher;
41+
import java.util.regex.Pattern;
42+
import java.util.stream.Collectors;
43+
44+
import static com.alibaba.otter.canal.deployer.CanalConstants.CANAL_DESTINATIONS;
45+
import static com.alibaba.otter.canal.deployer.CanalConstants.CANAL_DESTINATIONS_EXPR;
3946

4047
/**
4148
* canal调度控制器
@@ -390,7 +397,7 @@ private PlainCanalConfigClient getManagerClient(String managerAddress) {
390397
}
391398

392399
private void initInstanceConfig(Properties properties) {
393-
String destinationStr = getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
400+
String destinationStr = getDestinations(properties);
394401
String[] destinations = StringUtils.split(destinationStr, CanalConstants.CANAL_DESTINATION_SPLIT);
395402

396403
for (String destination : destinations) {
@@ -461,6 +468,40 @@ public static String getProperty(Properties properties, String key) {
461468
return StringUtils.trim(value);
462469
}
463470

471+
public static String getDestinations(Properties properties) {
472+
String expr = getProperty(properties, CANAL_DESTINATIONS_EXPR);
473+
if (StringUtils.isNotBlank(expr)) {
474+
return parseExpr(expr);
475+
} else {
476+
return getProperty(properties, CANAL_DESTINATIONS);
477+
}
478+
}
479+
480+
private static String parseExpr(String expr) {
481+
String prefix = StringUtils.substringBefore(expr, "{");
482+
String range = StringUtils.substringAfter(expr, "{");
483+
range = StringUtils.substringBefore(range, "}");
484+
485+
String regex = "(\\d+)-(\\d+)";
486+
Pattern pattern = Pattern.compile(regex);
487+
Matcher matcher = pattern.matcher(range);
488+
if (matcher.find()) {
489+
String head = matcher.group(1);
490+
String tail = matcher.group(2);
491+
int start = Integer.parseInt(head);
492+
int end = Integer.parseInt(tail);
493+
494+
List<String> list = new ArrayList<>();
495+
for (int i = start; i <= end; i++) {
496+
String d = prefix + i;
497+
list.add(d);
498+
}
499+
return list.stream().map(Object::toString).collect(Collectors.joining(","));
500+
} else {
501+
throw new CanalServerException("invalid destinations expr " + expr);
502+
}
503+
}
504+
464505
public void start() throws Throwable {
465506
logger.info("## start the canal server[{}({}):{}]", ip, registerIp, port);
466507
// 创建整个canal的工作节点

‎deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ public synchronized void start() throws Throwable {
102102

103103
if (canalMQProducer != null) {
104104
canalMQStarter = new CanalMQStarter(canalMQProducer);
105-
String destinations = CanalController.getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
105+
String destinations = CanalController.getDestinations(properties);
106106
canalMQStarter.start(destinations);
107107
controller.setCanalMQStarter(canalMQStarter);
108108
}

‎docker/image/app.sh

+36-14
Original file line numberDiff line numberDiff line change
@@ -95,24 +95,18 @@ function start_canal() {
9595
fi
9696

9797
destination=`perl -le 'print $ENV{"canal.destinations"}'`
98+
destinationExpr=`perl -le 'print $ENV{"canal.destinations.expr"}'`
9899
multistream=`perl -le 'print $ENV{"canal.instance.multi.stream.on"}'`
99-
if [[ "$destination" =~ ',' ]] ; then
100+
101+
if [[ "$destination" =~ ',' ]] || [[ -n "$destinationExpr" ]]; then
100102
if [[ "$multistream" = 'true' ]] ; then
101-
holdExample="false"
102-
array=(${destination//,/ })
103-
for var in ${array[@]}
104-
do
105-
cp -r /home/admin/canal-server/conf/example /home/admin/canal-server/conf/$var
106-
chown admin:admin -R /home/admin/canal-server/conf/$var
107-
if [[ "$var" = 'example' ]] ; then
108-
holdExample="true"
109-
fi
110-
done
111-
if [[ "$holdExample" != 'true' ]] ; then
112-
rm -rf /home/admin/canal-server/conf/example
103+
if [[ -n "$destinationExpr" ]] ; then
104+
splitDestinations '1' $destinationExpr
105+
else
106+
splitDestinations '2' $destination
113107
fi
114108
else
115-
echo "multi destination:$destination is not support"
109+
echo "multi destination is not support, destinationExpr:$destinationExpr, destinations:$destination"
116110
exit 1;
117111
fi
118112
else
@@ -130,6 +124,34 @@ function start_canal() {
130124
fi
131125
}
132126

127+
function splitDestinations() {
128+
holdExample="false"
129+
prefix=''
130+
array=()
131+
132+
if [[ "$1" == '1' ]] ; then
133+
echo "split destinations expr "$2
134+
prefix=$(echo $2 | sed 's/{.*//')
135+
num=$(echo $2 | sed 's/.*{//;s/}//;s/-/ /')
136+
array=($(seq $num))
137+
else
138+
echo "split destinations "$2
139+
array=(${2//,/ })
140+
fi
141+
142+
for var in ${array[@]}
143+
do
144+
cp -r /home/admin/canal-server/conf/example /home/admin/canal-server/conf/$prefix$var
145+
chown admin:admin -R /home/admin/canal-server/conf/$prefix$var
146+
if [[ "$prefix$var" = 'example' ]] ; then
147+
holdExample="true"
148+
fi
149+
done
150+
if [[ "$holdExample" != 'true' ]] ; then
151+
rm -rf /home/admin/canal-server/conf/example
152+
fi
153+
}
154+
133155
function stop_canal() {
134156
echo "stop canal"
135157
su admin -c 'cd /home/admin/canal-server/bin/ && sh stop.sh 1>>/tmp/start.log 2>&1'

0 commit comments

Comments
 (0)