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

Merge branch 'master' into metrics_support

Browse Source
This commit is contained in:
Chuanyi Li L
2018-08-06 15:26:44 +08:00
parent 8b186d1330 7c7ce43455
commit d01195ac82
28 changed files with 668 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -14,3 +14,5 @@ jtester.properties
.idea/
*.iml
.DS_Store
*.tar.gz
*.rpm

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

@@ -379,7 +379,18 @@ public class CanalController {
}
private String getProperty(Properties properties, String key) {
return StringUtils.trim(properties.getProperty(StringUtils.trim(key)));
key = StringUtils.trim(key);
String value = System.getProperty(key);
if (value == null) {
value = System.getenv(key);
}
if (value == null) {
value = properties.getProperty(key);
}
return StringUtils.trim(value);
}
public void start() throws Throwable {

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

@@ -3,7 +3,7 @@
#################################################
canal.id= 1
canal.ip=
canal.port= 11111
canal.port=11111
canal.zkServers=
# flush data to zk
canal.zookeeper.flush.period = 1000
@@ -59,6 +59,13 @@ canal.instance.parser.parallel = true
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256
# table meta tsdb info
canal.instance.tsdb.enable=true
canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername=canal
canal.instance.tsdb.dbPassword=canal
#################################################
######### destinations #############
#################################################

19
deployer/src/main/resources/example/instance.properties
View File

@@ -1,11 +1,12 @@
#################################################
## mysql serverId
canal.instance.mysql.slaveId=0
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0
# enable gtid use true/false
canal.instance.gtidon=false
# position info
canal.instance.master.address=127.0.0.1:3306
# enable gtid use true/false
canal.instance.gtidon=false
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
@@ -13,23 +14,21 @@ canal.instance.master.gtid=
# table meta tsdb info
canal.instance.tsdb.enable=true
canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
canal.instance.tsdb.dbUsername=canal
canal.instance.tsdb.dbPassword=canal
#canal.instance.tsdb.dbUsername=canal
#canal.instance.tsdb.dbPassword=canal
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.defaultDatabaseName=test
canal.instance.connectionCharset=UTF-8
# table regex
canal.instance.filter.regex=.*\\..*
# table black regex

4
deployer/src/main/resources/spring/default-instance.xml
View File

@@ -149,7 +149,7 @@
<property name="address" value="${canal.instance.master.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>
<property name="standbyInfo">
@@ -157,7 +157,7 @@
<property name="address" value="${canal.instance.standby.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>

4
deployer/src/main/resources/spring/file-instance.xml
View File

@@ -134,7 +134,7 @@
<property name="address" value="${canal.instance.master.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>
<property name="standbyInfo">
@@ -142,7 +142,7 @@
<property name="address" value="${canal.instance.standby.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>

8
deployer/src/main/resources/spring/group-instance.xml
View File

@@ -131,7 +131,7 @@
<property name="address" value="${canal.instance.master1.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>
<property name="standbyInfo">
@@ -139,7 +139,7 @@
<property name="address" value="${canal.instance.standby1.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>
@@ -229,7 +229,7 @@
<property name="address" value="${canal.instance.master2.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>
<property name="standbyInfo">
@@ -237,7 +237,7 @@
<property name="address" value="${canal.instance.standby2.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>

2
deployer/src/main/resources/spring/local-instance.xml
View File

@@ -114,7 +114,7 @@
<property name="address" value="${canal.instance.master.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>

4
deployer/src/main/resources/spring/memory-instance.xml
View File

@@ -122,7 +122,7 @@
<property name="address" value="${canal.instance.master.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>
<property name="standbyInfo">
@@ -130,7 +130,7 @@
<property name="address" value="${canal.instance.standby.address}" />
<property name="username" value="${canal.instance.dbUsername:retl}" />
<property name="password" value="${canal.instance.dbPassword:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:retl}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:test}" />
</bean>
</property>

67
docker/Dockerfile Normal file
View File

@@ -0,0 +1,67 @@
FROM centos:centos6.7
MAINTAINER agapple (jianghang115@gmail.com)
# install system
RUN \
/bin/cp -f /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \
echo 'root:Hello1234' | chpasswd && \
groupadd -r admin && useradd -g admin admin && \
yum install -y man && \
yum install -y dstat && \
yum install -y unzip && \
yum install -y nc && \
yum install -y openssh-server && \
yum install -y tar && \
yum install -y which && \
yum install -y wget && \
yum install -y perl && \
yum install -y file && \
ssh-keygen -q -N "" -t dsa -f /etc/ssh/ssh_host_dsa_key && \
ssh-keygen -q -N "" -t rsa -f /etc/ssh/ssh_host_rsa_key && \
sed -ri 's/session required pam_loginuid.so/#session required pam_loginuid.so/g' /etc/pam.d/sshd && \
sed -i -e 's/^#Port 22$/Port 2222/' /etc/ssh/sshd_config && \
mkdir -p /root/.ssh && chown root.root /root && chmod 700 /root/.ssh && \
yum install -y cronie && \
sed -i '/session required pam_loginuid.so/d' /etc/pam.d/crond && \
yum clean all && \
true
# install canal
COPY image/ /tmp/docker/
COPY canal.deployer-*.tar.gz /home/admin/
COPY jdk-8-linux-x64.rpm /tmp/
RUN \
cp -R /tmp/docker/alidata /alidata && \
chmod +x /alidata/bin/* && \
mkdir -p /home/admin && \
cp -R /tmp/docker/admin/* /home/admin/ && \
/bin/cp -f alidata/bin/lark-wait /usr/bin/lark-wait && \
touch /var/lib/rpm/* && \
yum -y install /tmp/jdk-8-linux-x64.rpm && \
/bin/rm -f /tmp/jdk-8-linux-x64.rpm && \
echo "export JAVA_HOME=/usr/java/latest" >> /etc/profile && \
echo "export PATH=\$JAVA_HOME/bin:\$PATH" >> /etc/profile && \
/bin/mv /home/admin/bin/clean_log /etc/cron.d && \
mkdir -p /home/admin/canal-server && \
tar -xzvf /home/admin/canal.deployer-*.tar.gz -C /home/admin/canal-server && \
/bin/rm -f /home/admin/canal.deployer-*.tar.gz && \
mkdir -p home/admin/canal-server/logs && \
chmod +x /home/admin/*.sh && \
chmod +x /home/admin/bin/*.sh && \
chown admin: -R /home/admin && \
yum clean all && \
true
# 2222 sys , 8080 web , 8000 debug , 11111 canal
EXPOSE 2222 11111 8000 8080
WORKDIR /home/admin
ENTRYPOINT [ "/alidata/bin/main.sh" ]
CMD [ "/home/admin/app.sh" ]

30
docker/build.sh Normal file
View File

@@ -0,0 +1,30 @@
#!/bin/bash
current_path=`pwd`
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
;;
*)
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
BASE=${bin_abs_path}
if [ ! -f $BASE/jdk*.rpm ] ; then
DOWNLOAD_LINK="http://download.oracle.com/otn-pub/java/jdk/8u181-b13/96a7b8442fe848ef90c96a2fad6ed6d1/jdk-8u181-linux-x64.tar.gz"
wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=xxx; oraclelicense=accept-securebackup-cookie" "$DOWNLOAD_LINK" -O $BASE/jdk-8-linux-x64.rpm
fi
cd $BASE/../ && mvn clean package -Dmaven.test.skip -Denv=release && cd $current_path ;
if [ "$1" == "kafka" ] ; then
cp $BASE/../target/canal-kafka-*.tar.gz $BASE/
docker build --no-cache -t canal/canal-server $BASE/
else
cp $BASE/../target/canal.deployer-*.tar.gz $BASE/
docker build --no-cache -t canal/canal-server $BASE/
fi

117
docker/image/admin/app.sh Executable file
View File

@@ -0,0 +1,117 @@
#!/bin/bash
set -e
source /etc/profile
export JAVA_HOME=/usr/java/latest
export PATH=$JAVA_HOME/bin:$PATH
touch /tmp/start.log
chown admin: /tmp/start.log
chown -R admin: /home/admin/canal-server
host=`hostname -i`
# waitterm
# wait TERM/INT signal.
# see: http://veithen.github.io/2014/11/16/sigterm-propagation.html
waitterm() {
local PID
# any process to block
tail -f /dev/null &
PID="$!"
# setup trap, could do nothing, or just kill the blocker
trap "kill -TERM ${PID}" TERM INT
# wait for signal, ignore wait exit code
wait "${PID}" || true
# clear trap
trap - TERM INT
# wait blocker, ignore blocker exit code
wait "${PID}" 2>/dev/null || true
}
# waittermpid "${PIDFILE}".
# monitor process by pidfile && wait TERM/INT signal.
# if the process disappeared, return 1, means exit with ERROR.
# if TERM or INT signal received, return 0, means OK to exit.
waittermpid() {
local PIDFILE PID do_run error
PIDFILE="${1?}"
do_run=true
error=0
trap "do_run=false" TERM INT
while "${do_run}" ; do
PID="$(cat "${PIDFILE}")"
if ! ps -p "${PID}" >/dev/null 2>&1 ; then
do_run=false
error=1
else
sleep 1
fi
done
trap - TERM INT
return "${error}"
}
function checkStart() {
local name=$1
local cmd=$2
local timeout=$3
cost=5
while [ $timeout -gt 0 ]; do
ST=`eval $cmd`
if [ "$ST" == "0" ]; then
sleep 1
let timeout=timeout-1
let cost=cost+1
elif [ "$ST" == "" ]; then
sleep 1
let timeout=timeout-1
let cost=cost+1
else
break
fi
done
echo "start $name successful"
}
function start_canal() {
echo "start canal ..."
serverPort=`perl -le 'print $ENV{"canal.port"}'`
if [ -z "$serverPort" ] ; then
serverPort=11111
fi
destination=`perl -le 'print $ENV{"canal.destinations"}'`
if [[ "$destination" =~ ',' ]]; then
echo "multi destination:$destination is not support"
exit 1;
else
mv /home/admin/canal-server/conf/example /home/admin/canal-server/conf/$destination
fi
su admin -c 'cd /home/admin/canal-server/bin/ && sh restart.sh 1>>/tmp/start.log 2>&1'
sleep 5
#check start
checkStart "canal" "nc 127.0.0.1 $serverPort -w 1 -z | wc -l" 30
}
function stop_canal() {
echo "stop canal"
su admin -c 'cd /home/admin/canal-server/bin/ && sh stop.sh 1>>/tmp/start.log 2>&1'
echo "stop canal successful ..."
}
echo "==> START ..."
start_canal
echo "==> START SUCCESSFUL ..."
tail -f /dev/null &
# wait TERM signal
waitterm
echo "==> STOP"
stop_canal
echo "==> STOP SUCCESSFUL ..."

2
docker/image/admin/bin/clean_log Normal file
View File

@@ -0,0 +1,2 @@
# cron clean log once per minute
*/2 * * * * admin /home/admin/bin/clean_log.sh >>/tmp/clean_log.log 2>&1

45
docker/image/admin/bin/clean_log.sh Executable file
View File

@@ -0,0 +1,45 @@
#!/bin/bash
# Global Settings
PATH="$HOME/bin:/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin:/usr/X11R6/bin:/root/bin"
export PATH
CUTOFF="85"
#获取磁盘使用率最高的分区
USAGE=$(df -h|awk 'NR>1 {gsub(/%$/,"",$5);print $5 }'|sort -nr|head -1)
before=$USAGE
baseClean(){
#删除tmp目录15天前的文件。
#更新文档时间戳
if [ -d /tmp/hsperfdata_admin ]
then
touch /tmp/hsperfdata_admin
touch /tmp/hsperfdata_admin/*
fi
find /tmp/ -type f -mtime +15 | xargs -t rm -rf >/dev/null 2>&1
now=$(df -h|awk 'NR>1 {gsub(/%$/,"",$5);print $5 }'|sort -nr|head -1)
echo "before:$before; now:$now"
}
CANAL_DIR="/home/admin/canal-server/logs"
if [[ -d $CANAL_DIR ]]; then
USAGE=$(df -h|awk 'NR>1 {gsub(/%$/,"",$5);print $5 }'|sort -nr|head -1)
if [[ $USAGE -ge 90 ]]; then
find $CANAL_DIR -type f -mtime +7 | xargs rm -rf {}
fi
USAGE=$(df -h|awk 'NR>1 {gsub(/%$/,"",$5);print $5 }'|sort -nr|head -1)
if [[ $USAGE -ge 80 ]]; then
find $CANAL_DIR -type f -mtime +3 | xargs rm -rf {}
fi
USAGE=$(df -h|awk 'NR>1 {gsub(/%$/,"",$5);print $5 }'|sort -nr|head -1)
if [[ $USAGE -ge 80 ]]; then
find $CANAL_DIR -type d -empty -mtime +3 | grep -v canal | xargs rm -rf {}
find $CANAL_DIR -type f -iname '*.tmp' | xargs rm -rf {}
fi
baseClean
exit 0
fi

13
docker/image/admin/health.sh Executable file
View File

@@ -0,0 +1,13 @@
#!/bin/sh
CHECK_URL="http://127.0.0.1:8080/metrics"
CHECK_POINT="success"
CHECK_COUNT=`curl -s --connect-timeout 7 --max-time 7 $CHECK_URL | grep -c $CHECK_POINT`
if [ $CHECK_COUNT -eq 0 ]; then
echo "[FAILED]"
status=0
error=1
else
echo "[ OK ]"
status=1
error=0
fi

11
docker/image/alidata/bin/exec_rc_local.sh Executable file
View File

@@ -0,0 +1,11 @@
#!/bin/bash
if [ "${SKIP_EXEC_RC_LOCAL}" = "YES" ] ; then
echo "skip /etc/rc.local: SKIP_EXEC_RC_LOCAL=${SKIP_EXEC_RC_LOCAL}"
exit
fi
if [ "${DOCKER_DEPLOY_TYPE}" = "HOST" ] ; then
echo "skip /etc/rc.local: DOCKER_DEPLOY_TYPE=${DOCKER_DEPLOY_TYPE}"
exit
fi

6
docker/image/alidata/bin/lark-wait Executable file
View File

@@ -0,0 +1,6 @@
#!/bin/bash
set -e
chown admin: -R /home/admin/
source /alidata/lib/proc.sh
waitterm

27
docker/image/alidata/bin/main.sh Executable file
View File

@@ -0,0 +1,27 @@
#!/bin/bash
[ -n "${DOCKER_DEPLOY_TYPE}" ] || DOCKER_DEPLOY_TYPE="VM"
echo "DOCKER_DEPLOY_TYPE=${DOCKER_DEPLOY_TYPE}"
# run init scripts
for e in $(ls /alidata/init/*) ; do
[ -x "${e}" ] || continue
echo "==> INIT $e"
$e
echo "==> EXIT CODE: $?"
done
echo "==> INIT DEFAULT"
service sshd start
service crond start
#echo "check hostname -i: `hostname -i`"
#hti_num=`hostname -i|awk '{print NF}'`
#if [ $hti_num -gt 1 ];then
# echo "hostname -i result error:`hostname -i`"
# exit 120
#fi
echo "==> INIT DONE"
echo "==> RUN ${*}"
exec "${@}"

19
docker/image/alidata/init/02init-sshd.sh Executable file
View File

@@ -0,0 +1,19 @@
#!/bin/bash
# set port
if [ -z "${SSHD_PORT}" ] ; then
SSHD_PORT=22
[ "${DOCKER_DEPLOY_TYPE}" = "HOST" ] && SSHD_PORT=2222
fi
sed -r -i '/^OPTIONS=/ d' /etc/sysconfig/sshd
echo 'OPTIONS="-p '"${SSHD_PORT}"'"' >> /etc/sysconfig/sshd
# set admin ssh pulic key
if [ "${USE_ADMIN_PASSAGE}" = "YES" ] ; then
echo "set admin passage"
mkdir -p /home/admin/.ssh
chown admin:admin /home/admin/.ssh
chown admin:admin /home/admin/.ssh/authorized_keys
chmod 644 /home/admin/.ssh/authorized_keys
fi

66
docker/image/alidata/init/fix-hosts.py Executable file
View File

@@ -0,0 +1,66 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#****************************************************************#
# Create Date: 2017-01-06 17:58
#***************************************************************#
import socket
import shutil
from time import gmtime, strftime
# get host_name
host_name = socket.gethostname()
tmp_file = "/tmp/.lark-fix-host.hosts"
host_file = "/etc/hosts"
bak_file_name = "/tmp/hosts-fix-bak.%s" % ( strftime("%Y-%m-%d_%H-%M-%S", gmtime()) )
# load /etc/hosts file context
FH = open(host_file,"r")
file_lines = [ i.rstrip() for i in FH.readlines()]
FH.close()
file_lines_reverse = file_lines[::-1]
new_lines = []
bad_lines = []
last_match_line = ""
for line in file_lines_reverse:
if line.find(host_name) < 0: # 不匹配的行直接跳过
new_lines.append(line + "\n")
continue
cols = line.split()
new_cols = []
if cols[0].startswith("#"): # 跳过已经注释掉的行
new_lines.append(line + "\n")
continue
for col in cols:
if not col == host_name: # 跳过不匹配的列
new_cols.append(col)
continue
if cols[0] == "127.0.0.1": # 如果第一列是 127.0.0.1 就跳过匹配的列, 防止 hostname -i 返回 127.0.0.1
continue
# 如果已经发现过匹配的列, 就丢掉重复的列
if not len(last_match_line) == 0:
continue
new_cols.append(col)
last_match_line = line
# 跳过 xx.xx.xx.xx hostname 这样的重复列
if len(new_cols) == 1:
continue
new_l = "%s\n" % " ".join(new_cols)
new_lines.append(new_l)
# save tmp hosts
FH2=file(tmp_file,"w+")
FH2.writelines( new_lines[::-1])
FH2.close()
# mv to /etc/hosts
shutil.copy(host_file, bak_file_name)
shutil.move(tmp_file, host_file)

40
docker/image/alidata/lib/proc.sh Normal file
View File

@@ -0,0 +1,40 @@
# waitterm
# wait TERM/INT signal.
# see: http://veithen.github.io/2014/11/16/sigterm-propagation.html
waitterm() {
local PID
# any process to block
tail -f /dev/null &
PID="$!"
# setup trap, could do nothing, or just kill the blocker
trap "kill -TERM ${PID}" TERM INT
# wait for signal, ignore wait exit code
wait "${PID}" || true
# clear trap
trap - TERM INT
# wait blocker, ignore blocker exit code
wait "${PID}" 2>/dev/null || true
}
# waittermpid "${PIDFILE}".
# monitor process by pidfile && wait TERM/INT signal.
# if the process disappeared, return 1, means exit with ERROR.
# if TERM or INT signal received, return 0, means OK to exit.
waittermpid() {
local PIDFILE PID do_run error
PIDFILE="${1?}"
do_run=true
error=0
trap "do_run=false" TERM INT
while "${do_run}" ; do
PID="$(cat "${PIDFILE}")"
if ! ps -p "${PID}" >/dev/null 2>&1 ; then
do_run=false
error=1
else
sleep 1
fi
done
trap - TERM INT
return "${error}"
}

92
docker/run.sh Normal file
View File

@@ -0,0 +1,92 @@
#!/bin/bash
function usage() {
echo "Usage:"
echo " run.sh [CONFIG]"
echo "example:"
echo " run.sh -e canal.instance.master.address=127.0.0.1:3306 \\"
echo " -e canal.instance.dbUsername=canal \\"
echo " -e canal.instance.dbPassword=canal \\"
echo " -e canal.instance.connectionCharset=UTF-8 \\"
echo " -e canal.instance.tsdb.enable=true \\"
echo " -e canal.instance.gtidon=false \\"
echo " -e canal.instance.filter.regex=.*\\..* "
exit
}
function check_port() {
local port=$1
local TL=$(which telnet)
if [ -f $TL ]; then
data=`echo quit | telnet 127.0.0.1 $port| grep -ic connected`
echo $data
return
fi
local NC=$(which nc)
if [ -f $NC ]; then
data=`nc -z -w 1 127.0.0.1 $port | grep -ic succeeded`
echo $data
return
fi
echo "0"
return
}
function getMyIp() {
case "`uname`" in
Darwin)
myip=`echo "show State:/Network/Global/IPv4" | scutil | grep PrimaryInterface | awk '{print $3}' | xargs ifconfig | grep inet | grep -v inet6 | awk '{print $2}'`
;;
*)
myip=`ip route get 1 | awk '{print $NF;exit}'`
;;
esac
echo $myip
}
NET_MODE=""
case "`uname`" in
Darwin)
bin_abs_path=`cd $(dirname $0); pwd`
;;
Linux)
bin_abs_path=$(readlink -f $(dirname $0))
NET_MODE="--net=host"
;;
*)
NET_MODE="--net=host"
bin_abs_path=`cd $(dirname $0); pwd`
;;
esac
BASE=${bin_abs_path}
if [ $# -eq 0 ]; then
usage
elif [ "$1" == "-h" ] ; then
usage
elif [ "$1" == "help" ] ; then
usage
fi
DATA="$BASE/data"
mkdir -p $DATA
CONFIG=${@:1}
#VOLUMNS="-v $DATA:/home/admin/canal-server/logs"
PORTLIST="8000 8080 2222 11111"
PORTS=""
for PORT in $PORTLIST ; do
#exist=`check_port $PORT`
exist="0"
if [ "$exist" == "0" ]; then
PORTS="$PORTS -p $PORT:$PORT"
else
echo "port $PORT is used , pls check"
exit 1
fi
done
MEMORY="-m 4096m"
LOCALHOST=`getMyIp`
cmd="docker run -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
echo $cmd
eval $cmd

2
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java
View File

@@ -32,7 +32,7 @@ public class MysqlConnector {
private String password;
private byte charsetNumber = 33;
private String defaultSchema = "retl";
private String defaultSchema = "test";
private int soTimeout = 30 * 1000;
private int connTimeout = 5 * 1000;
private int receiveBufferSize = 16 * 1024;

21
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java
View File

@@ -6,6 +6,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import com.alibaba.otter.canal.meta.FileMixedMetaManager;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -20,13 +21,7 @@ import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.DataSourcing;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.HAMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.IndexMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageScavengeMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.*;
import com.alibaba.otter.canal.meta.MemoryMetaManager;
import com.alibaba.otter.canal.meta.PeriodMixedMetaManager;
import com.alibaba.otter.canal.meta.ZooKeeperMetaManager;
@@ -37,12 +32,7 @@ import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.index.FailbackLogPositionManager;
import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;
import com.alibaba.otter.canal.parse.index.MetaLogPositionManager;
import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager;
import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager;
import com.alibaba.otter.canal.parse.index.*;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.sink.entry.EntryEventSink;
@@ -120,6 +110,11 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
ZooKeeperMetaManager zooKeeperMetaManager = new ZooKeeperMetaManager();
zooKeeperMetaManager.setZkClientx(getZkclientx());
((PeriodMixedMetaManager) metaManager).setZooKeeperMetaManager(zooKeeperMetaManager);
} else if (mode.isLocalFile()){
FileMixedMetaManager fileMixedMetaManager = new FileMixedMetaManager();
fileMixedMetaManager.setDataDir(parameters.getDataDir());
fileMixedMetaManager.setPeriod(parameters.getMetaFileFlushPeriod());
metaManager = fileMixedMetaManager;
} else {
throw new CanalException("unsupport MetaMode for " + mode);
}

26
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java
View File

@@ -28,8 +28,10 @@ public class CanalParameter implements Serializable {
private Long zkClusterId; // zk集群id，为管理方便
private List<String> zkClusters; // zk集群地址
private String dataDir = "../conf"; // 默认本地文件数据的目录默认是conf
// meta相关参数
private MetaMode metaMode = MetaMode.MEMORY; // meta机制
private Integer metaFileFlushPeriod = 1000; // meta刷新间隔
// storage存储
private Integer transactionSize = 1024; // 支持处理的transaction事务大小
@@ -243,7 +245,9 @@ public class CanalParameter implements Serializable {
/** 文件存储模式 */
ZOOKEEPER,
/** 混合模式，内存+文件 */
MIXED;
MIXED,
/** 本地文件存储模式*/
LOCAL_FILE;
public boolean isMemory() {
return this.equals(MetaMode.MEMORY);
@@ -256,6 +260,10 @@ public class CanalParameter implements Serializable {
public boolean isMixed() {
return this.equals(MetaMode.MIXED);
}
public boolean isLocalFile(){
return this.equals(MetaMode.LOCAL_FILE);
}
}
public static enum IndexMode {
@@ -390,6 +398,22 @@ public class CanalParameter implements Serializable {
return storageMode;
}
public String getDataDir() {
return dataDir;
}
public void setDataDir(String dataDir) {
this.dataDir = dataDir;
}
public Integer getMetaFileFlushPeriod() {
return metaFileFlushPeriod;
}
public void setMetaFileFlushPeriod(Integer metaFileFlushPeriod) {
this.metaFileFlushPeriod = metaFileFlushPeriod;
}
public void setStorageMode(StorageMode storageMode) {
this.storageMode = storageMode;
}

2
kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java
View File

@@ -121,7 +121,7 @@ public class CanalKafkaStarter {
Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据
long batchId = message.getId();
try {
int size = message.getEntries().size();
int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size();
if (batchId != -1 && size != 0) {
if (!StringUtils.isEmpty(destination.getTopic())) {
Topic topic = new Topic();

3
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
View File

@@ -648,6 +648,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
throw new CanalParseException("command : 'show master status' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
}
EntryPosition endPosition = new EntryPosition(fields.get(0), Long.valueOf(fields.get(1)));
if (isGTIDMode && fields.size() > 4) {
endPosition.setGtid(fields.get(4));
}
return endPosition;
} catch (IOException e) {
throw new CanalParseException("command : 'show master status' has an error!", e);

53
server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithEmbedded_FileModeTest.java Normal file
View File

@@ -0,0 +1,53 @@
package com.alibaba.otter.canal.server;
import java.net.InetSocketAddress;
import java.util.Arrays;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.*;
public class CanalServerWithEmbedded_FileModeTest extends BaseCanalServerWithEmbededTest {
protected Canal buildCanal() {
Canal canal = new Canal();
canal.setId(1L);
canal.setName(DESTINATION);
canal.setDesc("my standalone server test ");
CanalParameter parameter = new CanalParameter();
parameter.setMetaMode(MetaMode.LOCAL_FILE);
parameter.setDataDir("./conf");
parameter.setMetaFileFlushPeriod(1000);
parameter.setHaMode(HAMode.HEARTBEAT);
parameter.setIndexMode(IndexMode.MEMORY_META_FAILBACK);
parameter.setStorageMode(StorageMode.MEMORY);
parameter.setMemoryStorageBufferSize(32 * 1024);
parameter.setSourcingType(SourcingType.MYSQL);
parameter.setDbAddresses(Arrays.asList(new InetSocketAddress(MYSQL_ADDRESS, 3306),
new InetSocketAddress(MYSQL_ADDRESS, 3306)));
parameter.setDbUsername(USERNAME);
parameter.setDbPassword(PASSWORD);
parameter.setPositions(Arrays.asList("{\"journalName\":\"mysql-bin.000001\",\"position\":332L,\"timestamp\":\"1505998863000\"}",
"{\"journalName\":\"mysql-bin.000001\",\"position\":332L,\"timestamp\":\"1505998863000\"}"));
parameter.setSlaveId(1234L);
parameter.setDefaultConnectionTimeoutInSeconds(30);
parameter.setConnectionCharset("UTF-8");
parameter.setConnectionCharsetNumber((byte) 33);
parameter.setReceiveBufferSize(8 * 1024);
parameter.setSendBufferSize(8 * 1024);
parameter.setDetectingEnable(false);
parameter.setDetectingIntervalInSeconds(10);
parameter.setDetectingRetryTimes(3);
parameter.setDetectingSQL(DETECTING_SQL);
canal.setCanalParameter(parameter);
return canal;
}
}