From d3f52c6d3d683dd4865ae2ac83431982f3d2a5e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Mon, 13 Aug 2018 10:16:28 +0800 Subject: [PATCH 1/9] fixed issue #727 , support rds binlog dump --- .../tddl/dbsync/binlog/FileLogFetcher.java | 4 +- deployer/src/main/resources/canal.properties | 5 +- .../resources/example/instance.properties | 5 + .../resources/example/rds_instance.properties | 32 -- .../main/resources/spring/base-instance.xml | 39 +++ .../resources/spring/default-instance.xml | 23 +- .../main/resources/spring/file-instance.xml | 23 +- .../main/resources/spring/group-instance.xml | 25 +- .../main/resources/spring/local-instance.xml | 147 -------- .../main/resources/spring/memory-instance.xml | 23 +- .../exception/ServerIdNotMatchException.java | 2 + .../parse/inbound/AbstractEventParser.java | 14 +- .../parse/inbound/MultiStageCoprocessor.java | 3 +- .../mysql/AbstractMysqlEventParser.java | 8 +- .../inbound/mysql/LocalBinLogConnection.java | 57 +-- .../parse/inbound/mysql/MysqlConnection.java | 25 +- .../parse/inbound/mysql/MysqlEventParser.java | 47 +-- .../mysql/MysqlMultiStageCoprocessor.java | 57 +-- .../inbound/mysql/dbsync/LogEventConvert.java | 22 +- .../inbound/mysql/local/BinLogFileQueue.java | 62 +++- .../mysql/rds/BinlogDownloadQueue.java | 99 +++++- .../parse/inbound/mysql/rds/HttpHelper.java | 1 + .../mysql/rds/RdsBinlogEventParserProxy.java | 161 +++++---- .../inbound/mysql/rds/RdsBinlogOpenApi.java | 330 ++---------------- .../mysql/rds/RdsLocalBinlogEventParser.java | 33 +- .../inbound/mysql/rds/data/BinlogFile.java | 10 +- .../rds/data/DescribeBinlogFileResult.java | 11 +- .../mysql/rds/data/RdsBackupPolicy.java | 19 +- .../parse/inbound/mysql/rds/data/RdsItem.java | 7 + .../mysql/rds/request/AbstractRequest.java | 69 ++-- .../request/DescribeBackupPolicyRequest.java | 7 +- .../request/DescribeBinlogFilesRequest.java | 15 +- .../inbound/mysql/tsdb/MemoryTableMeta.java | 3 +- .../mysql/RdsBinlogEventParserProxyTest.java | 21 +- .../inbound/mysql/RdsBinlogOpenApiTest.java | 27 +- 35 files changed, 553 insertions(+), 883 deletions(-) delete mode 100644 deployer/src/main/resources/example/rds_instance.properties create mode 100644 deployer/src/main/resources/spring/base-instance.xml delete mode 100644 deployer/src/main/resources/spring/local-instance.xml diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/FileLogFetcher.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/FileLogFetcher.java index 0d586cd4..cdb1f685 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/FileLogFetcher.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/FileLogFetcher.java @@ -168,7 +168,9 @@ public final class FileLogFetcher extends LogFetcher { * @see com.taobao.tddl.dbsync.binlog.LogFetcher#close() */ public void close() throws IOException { - if (fin != null) fin.close(); + if (fin != null) { + fin.close(); + } fin = null; } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 2bff29ae..9083477d 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -66,6 +66,10 @@ canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MO canal.instance.tsdb.dbUsername=canal canal.instance.tsdb.dbPassword=canal +# rds oss binlog account +canal.instance.rds.accesskey = +canal.instance.rds.secretkey = + ################################################# ######### destinations ############# ################################################# @@ -82,7 +86,6 @@ 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.spring.xml = classpath:spring/local-instance.xml #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/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index e0cf0064..f72b65fe 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -12,6 +12,11 @@ canal.instance.master.position= canal.instance.master.timestamp= canal.instance.master.gtid= +# rds oss binlog +canal.instance.rds.accesskey= +canal.instance.rds.secretkey= +canal.instance.rds.instanceId= + # table meta tsdb info canal.instance.tsdb.enable=true #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb diff --git a/deployer/src/main/resources/example/rds_instance.properties b/deployer/src/main/resources/example/rds_instance.properties deleted file mode 100644 index f04b3832..00000000 --- a/deployer/src/main/resources/example/rds_instance.properties +++ /dev/null @@ -1,32 +0,0 @@ -################################################# -# rds openapi binlog -canal.instance.rds.open.url=https://rds.aliyuncs.com/ -canal.instance.rds.open.accesskey= -canal.instance.rds.open.secretkey= -canal.instance.rds.instanceId= -canal.instance.rds.startTime= -canal.instance.rds.endTime= - -# local binlog dir -canal.instance.parser.directory=${canal.file.data.dir:../conf}/${canal.instance.destination:}/binlog -# position info -canal.instance.master.address=127.0.0.1:3306 - -# 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 - -# 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=test\\..* -# table black regex -canal.instance.filter.black.regex= -################################################# \ No newline at end of file diff --git a/deployer/src/main/resources/spring/base-instance.xml b/deployer/src/main/resources/spring/base-instance.xml new file mode 100644 index 00000000..e6c500b5 --- /dev/null +++ b/deployer/src/main/resources/spring/base-instance.xml @@ -0,0 +1,39 @@ + + + + + + + + + + classpath:canal.properties + classpath:${canal.instance.destination:}/instance.properties + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index 6b455c83..07b9109d 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -10,26 +10,7 @@ http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd" default-autowire="byName"> - - - - - - - classpath:canal.properties - classpath:${canal.instance.destination:}/instance.properties - - - - - - - - - - - - + @@ -84,7 +65,7 @@ - + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 377428a9..58dd32c9 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -10,26 +10,7 @@ http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd" default-autowire="byName"> - - - - - - - classpath:canal.properties - classpath:${canal.instance.destination:}/instance.properties - - - - - - - - - - - - + @@ -70,7 +51,7 @@ - + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index 44f1ab3b..47d3ba33 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -10,26 +10,7 @@ http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd" default-autowire="byName"> - - - - - - - classpath:canal.properties - classpath:${canal.instance.destination:}/instance.properties - - - - - - - - - - - - + @@ -76,7 +57,7 @@ - + @@ -174,7 +155,7 @@ - + diff --git a/deployer/src/main/resources/spring/local-instance.xml b/deployer/src/main/resources/spring/local-instance.xml deleted file mode 100644 index 930ecb7b..00000000 --- a/deployer/src/main/resources/spring/local-instance.xml +++ /dev/null @@ -1,147 +0,0 @@ - - - - - - - - - - classpath:canal.properties - classpath:${canal.instance.destination:}/instance.properties - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index cd387b8f..428454db 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -10,26 +10,7 @@ http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd" default-autowire="byName"> - - - - - - - classpath:canal.properties - classpath:${canal.instance.destination:}/instance.properties - - - - - - - - - - - - + @@ -67,7 +48,7 @@ - + diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java index 35a6b790..05baee02 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java @@ -8,6 +8,8 @@ import com.alibaba.otter.canal.common.CanalException; */ public class ServerIdNotMatchException extends CanalException{ + private static final long serialVersionUID = -6124989280379293953L; + public ServerIdNotMatchException(String errorCode) { super(errorCode); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index cd50afa4..fcae1317 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -8,8 +8,6 @@ import java.util.TimerTask; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; - import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.RandomUtils; @@ -23,6 +21,7 @@ import com.alibaba.otter.canal.filter.CanalEventFilter; import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException; import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback; import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; @@ -97,10 +96,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle protected int parallelBufferSize = 256; // 必须为2的幂 protected MultiStageCoprocessor multiStageCoprocessor; protected ParserExceptionHandler parserExceptionHandler; - protected long serverId; - - - + protected long serverId; protected abstract BinlogParser buildParser(); @@ -179,7 +175,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle erosaConnection.connect();// 链接 long queryServerId = erosaConnection.queryServerId(); - if (queryServerId != 0){ + if (queryServerId != 0) { serverId = queryServerId; } // 4. 获取最后的位置信息 @@ -289,7 +285,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle runningInfo.getAddress().toString()), e); sendAlarm(destination, ExceptionUtils.getFullStackTrace(e)); } - if (parserExceptionHandler!=null){ + if (parserExceptionHandler != null) { parserExceptionHandler.handle(e); } } finally { @@ -441,7 +437,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle return logPosition; } - protected void processSinkError(Throwable e, LogPosition lastPosition, String startBinlogFile, long startPosition) { + protected void processSinkError(Throwable e, LogPosition lastPosition, String startBinlogFile, Long startPosition) { if (lastPosition != null) { logger.warn(String.format("ERROR ## parse this event has an error , last position : [%s]", lastPosition.getPostion()), diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java index 96e063e9..4ab68e39 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.inbound; import com.alibaba.otter.canal.common.CanalLifeCycle; import com.taobao.tddl.dbsync.binlog.LogBuffer; +import com.taobao.tddl.dbsync.binlog.LogEvent; /** * 针对解析器提供一个多阶段协同的处理 @@ -23,7 +24,7 @@ public interface MultiStageCoprocessor extends CanalLifeCycle { */ public boolean publish(LogBuffer buffer); - public boolean publish(LogBuffer buffer, String binlogFileName); + public boolean publish(LogEvent event); public void reset(); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index db21a9e1..941f5ebc 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -134,10 +134,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected MultiStageCoprocessor buildMultiStageCoprocessor() { MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize, - parallelThreadSize, - (LogEventConvert) binlogParser, - transactionBuffer, - destination); + parallelThreadSize, + (LogEventConvert) binlogParser, + transactionBuffer, + destination); mysqlMultiStageCoprocessor.setEventsPublishBlockingTime(eventsPublishBlockingTime); return mysqlMultiStageCoprocessor; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java index 0358dfda..5cf5bb9d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java @@ -4,25 +4,24 @@ import java.io.File; import java.io.IOException; import java.util.List; -import com.alibaba.otter.canal.parse.exception.ServerIdNotMatchException; import org.apache.commons.lang.NotImplementedException; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; +import com.alibaba.otter.canal.parse.exception.CanalParseException; +import com.alibaba.otter.canal.parse.exception.ServerIdNotMatchException; import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor; import com.alibaba.otter.canal.parse.inbound.SinkFunction; import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue; import com.taobao.tddl.dbsync.binlog.FileLogFetcher; -import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; import com.taobao.tddl.dbsync.binlog.LogPosition; import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent; /** * local bin log connection (not real connection) @@ -40,7 +39,6 @@ public class LocalBinLogConnection implements ErosaConnection { private long serverId; private FileParserListener parserListener; - public LocalBinLogConnection(){ } @@ -92,16 +90,13 @@ public class LocalBinLogConnection implements ErosaConnection { while (running) { boolean needContinue = true; LogEvent event = null; - // 处理一下binlog文件名 - event = new RotateLogEvent(context.getLogPosition().getFileName(), 4); - func.sink(event); while (fetcher.fetch()) { event = decoder.decode(fetcher, context); if (event == null) { continue; } - if (serverId != 0 && event.getServerId() != serverId){ - throw new ServerIdNotMatchException("unexpected serverId "+serverId + " in binlog file !"); + if (serverId != 0 && event.getServerId() != serverId) { + throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); } if (!func.sink(event)) { @@ -168,9 +163,8 @@ public class LocalBinLogConnection implements ErosaConnection { while (fetcher.fetch()) { LogEvent event = decoder.decode(fetcher, context); if (event != null) { - - if (serverId != 0 && event.getServerId() != serverId){ - throw new ServerIdNotMatchException("unexpected serverId "+serverId + " in binlog file !"); + if (serverId != 0 && event.getServerId() != serverId) { + throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); } if (event.getWhen() > timestampSeconds) { @@ -189,6 +183,9 @@ public class LocalBinLogConnection implements ErosaConnection { } else if (LogEvent.XID_EVENT == event.getHeader().getType()) { lastXidLogFilename = current.getName(); lastXidLogFileOffset = event.getLogPos(); + } else if (LogEvent.FORMAT_DESCRIPTION_EVENT == event.getHeader().getType()) { + lastXidLogFilename = current.getName(); + lastXidLogFileOffset = event.getLogPos(); } } } @@ -225,17 +222,29 @@ public class LocalBinLogConnection implements ErosaConnection { @Override public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException { File current = new File(directory, binlogfilename); + if (!current.exists()) { + throw new CanalParseException("binlog:" + binlogfilename + " is not found"); + } FileLogFetcher fetcher = new FileLogFetcher(bufferSize); + LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); + LogContext context = new LogContext(); try { fetcher.open(current, binlogPosition); + context.setLogPosition(new LogPosition(binlogfilename, binlogPosition)); while (running) { boolean needContinue = true; + LogEvent event = null; while (fetcher.fetch()) { - LogBuffer buffer = fetcher.duplicate(); - fetcher.consume(fetcher.limit()); - // set filename - if (!coprocessor.publish(buffer, binlogfilename)) { + event = decoder.decode(fetcher, context); + if (event == null) { + continue; + } + if (serverId != 0 && event.getServerId() != serverId) { + throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); + } + + if (!coprocessor.publish(event)) { needContinue = false; break; } @@ -244,7 +253,6 @@ public class LocalBinLogConnection implements ErosaConnection { fetcher.close(); // 关闭上一个文件 parserFinish(binlogfilename); if (needContinue) {// 读取下一个 - File nextFile; if (needWait) { nextFile = binlogs.waitForNextFile(current); @@ -272,8 +280,8 @@ public class LocalBinLogConnection implements ErosaConnection { } } - private void parserFinish(String fileName){ - if (parserListener != null){ + private void parserFinish(String fileName) { + if (parserListener != null) { parserListener.onFinish(fileName); } } @@ -306,9 +314,8 @@ public class LocalBinLogConnection implements ErosaConnection { while (fetcher.fetch()) { LogEvent event = decoder.decode(fetcher, context); if (event != null) { - - if (serverId != 0 && event.getServerId() != serverId){ - throw new ServerIdNotMatchException("unexpected serverId "+serverId + " in binlog file !"); + if (serverId != 0 && event.getServerId() != serverId) { + throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); } if (event.getWhen() > timestampSeconds) { @@ -327,6 +334,9 @@ public class LocalBinLogConnection implements ErosaConnection { } else if (LogEvent.XID_EVENT == event.getHeader().getType()) { lastXidLogFilename = current.getName(); lastXidLogFileOffset = event.getLogPos(); + } else if (LogEvent.FORMAT_DESCRIPTION_EVENT == event.getHeader().getType()) { + lastXidLogFilename = current.getName(); + lastXidLogFileOffset = event.getLogPos(); } } } @@ -410,7 +420,8 @@ public class LocalBinLogConnection implements ErosaConnection { this.parserListener = parserListener; } - public interface FileParserListener{ + public interface FileParserListener { + void onFinish(String fileName); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 54483e1f..eb3c0e31 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -13,7 +13,6 @@ import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.math.NumberUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.util.CollectionUtils; import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector; import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor; @@ -40,20 +39,20 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; public class MysqlConnection implements ErosaConnection { - private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); + private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); - private MysqlConnector connector; - private long slaveId; - private Charset charset = Charset.forName("UTF-8"); - private BinlogFormat binlogFormat; - private BinlogImage binlogImage; + private MysqlConnector connector; + private long slaveId; + private Charset charset = Charset.forName("UTF-8"); + private BinlogFormat binlogFormat; + private BinlogImage binlogImage; // tsdb releated - private AuthenticationInfo authInfo; - protected int connTimeout = 5 * 1000; // 5秒 - protected int soTimeout = 60 * 60 * 1000; // 1小时 + private AuthenticationInfo authInfo; + protected int connTimeout = 5 * 1000; // 5秒 + protected int soTimeout = 60 * 60 * 1000; // 1小时 // dump binlog bytes, 暂不包括meta与TSDB - private AtomicLong receivedBinlogBytes; + private AtomicLong receivedBinlogBytes; public MysqlConnection(){ } @@ -338,7 +337,7 @@ public class MysqlConnection implements ErosaConnection { public long queryServerId() throws IOException { ResultSetPacket resultSetPacket = query("show variables like 'server_id'"); List fieldValues = resultSetPacket.getFieldValues(); - if (fieldValues == null || fieldValues.size() != 2){ + if (fieldValues == null || fieldValues.size() != 2) { return 0; } return NumberUtils.toLong(fieldValues.get(1)); @@ -478,8 +477,6 @@ public class MysqlConnection implements ErosaConnection { } } - - public static enum BinlogFormat { STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED"); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index fefb1cbe..b585d9ab 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -47,32 +47,32 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; */ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalEventParser, CanalHASwitchable { - private CanalHAController haController = null; + private CanalHAController haController = null; - private int defaultConnectionTimeoutInSeconds = 30; // sotimeout - private int receiveBufferSize = 64 * 1024; - private int sendBufferSize = 64 * 1024; + private int defaultConnectionTimeoutInSeconds = 30; // sotimeout + private int receiveBufferSize = 64 * 1024; + private int sendBufferSize = 64 * 1024; // 数据库信息 - protected AuthenticationInfo masterInfo; // 主库 - protected AuthenticationInfo standbyInfo; // 备库 + protected AuthenticationInfo masterInfo; // 主库 + protected AuthenticationInfo standbyInfo; // 备库 // binlog信息 protected EntryPosition masterPosition; protected EntryPosition standbyPosition; - private long slaveId; // 链接到mysql的slave + private long slaveId; // 链接到mysql的slave // 心跳检查信息 - private String detectingSQL; // 心跳sql - private MysqlConnection metaConnection; // 查询meta信息的链接 - private TableMetaCache tableMetaCache; // 对应meta - private int fallbackIntervalInSeconds = 60; // 切换回退时间 - private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验 - private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验 + private String detectingSQL; // 心跳sql + private MysqlConnection metaConnection; // 查询meta信息的链接 + private TableMetaCache tableMetaCache; // 对应meta + private int fallbackIntervalInSeconds = 60; // 切换回退时间 + private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验 + private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验 // update by yishun.chen,特殊异常处理参数 - private int dumpErrorCount = 0; // binlogDump失败异常计数 - private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 + private int dumpErrorCount = 0; // binlogDump失败异常计数 + private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 // instance received binlog bytes - private final AtomicLong receivedBinlogBytes = new AtomicLong(0L); + private final AtomicLong receivedBinlogBytes = new AtomicLong(0L); protected ErosaConnection buildErosaConnection() { return buildMysqlConnection(this.runningInfo); @@ -756,16 +756,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE Long logposTimestamp = entry.getHeader().getExecuteTime(); Long serverId = entry.getHeader().getServerId(); - if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType()) - || CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) { - if (logger.isDebugEnabled()) { - logger.debug("compare exit condition:{},{},{}, startTimestamp={}...", new Object[] { - logfilename, logfileoffset, logposTimestamp, startTimestamp }); - } - // 事务头和尾寻找第一条记录时间戳，如果最小的一条记录都不满足条件，可直接退出 - if (logposTimestamp >= startTimestamp) { - return false; - } + // 如果最小的一条记录都不满足条件，可直接退出 + if (logposTimestamp >= startTimestamp) { + return false; } if (StringUtils.equals(endPosition.getJournalName(), logfilename) @@ -914,8 +907,6 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE this.dumpErrorCountThreshold = dumpErrorCountThreshold; } - - public AtomicLong getReceivedBinlogBytes() { return this.receivedBinlogBytes; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 31dfc595..63b0f0eb 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -5,8 +5,6 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.LockSupport; -import org.apache.commons.lang.StringUtils; - import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import com.alibaba.otter.canal.parse.exception.CanalParseException; @@ -32,7 +30,6 @@ import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; -import com.taobao.tddl.dbsync.binlog.LogPosition; import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent; @@ -53,7 +50,7 @@ import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; */ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implements MultiStageCoprocessor { - private static final int maxFullTimes = 10; + private static final int maxFullTimes = 10; private LogEventConvert logEventConvert; private EventTransactionBuffer transactionBuffer; private ErosaConnection connection; @@ -141,14 +138,18 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement super.stop(); } + public boolean publish(LogBuffer buffer) { + return this.publish(buffer, null); + } + /** * 网络数据投递 */ - public boolean publish(LogBuffer buffer) { - return publish(buffer, null); + public boolean publish(LogEvent event) { + return this.publish(null, event); } - public boolean publish(LogBuffer buffer, String binlogFileName) { + private boolean publish(LogBuffer buffer, LogEvent event) { if (!isStart()) { if (exception != null) { throw exception; @@ -169,10 +170,11 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement do { try { long next = disruptorMsgBuffer.tryNext(); - MessageEvent event = disruptorMsgBuffer.get(next); - event.setBuffer(buffer); - if (binlogFileName != null) { - event.setBinlogFileName(binlogFileName); + MessageEvent data = disruptorMsgBuffer.get(next); + if (buffer != null) { + data.setBuffer(buffer); + } else { + data.setEvent(event); } disruptorMsgBuffer.publish(next); if (fullTimes > 0) { @@ -184,7 +186,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement blockingStart = System.nanoTime(); } // park - //LockSupport.parkNanos(1L); + // LockSupport.parkNanos(1L); applyWait(++fullTimes); interupted = Thread.interrupted(); if (fullTimes % 1000 == 0) { @@ -217,7 +219,6 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement start(); } - private class SimpleParserStage implements EventHandler, LifecycleAware { private LogDecoder decoder; @@ -230,23 +231,13 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) throws Exception { try { - LogBuffer buffer = event.getBuffer(); - if (StringUtils.isNotEmpty(event.getBinlogFileName()) - && (context.getLogPosition() == null || !context.getLogPosition() - .getFileName() - .equals(event.getBinlogFileName()))) { - // set roate binlog file name - if (context.getLogPosition() == null) { - context.setLogPosition(new LogPosition(event.getBinlogFileName(), 0)); - } else { - context.setLogPosition(new LogPosition(event.getBinlogFileName(), context.getLogPosition() - .getPosition())); - } + LogEvent logEvent = event.getEvent(); + if (logEvent == null) { + LogBuffer buffer = event.getBuffer(); + logEvent = decoder.decode(buffer, context); + event.setEvent(logEvent); } - LogEvent logEvent = decoder.decode(buffer, context); - event.setEvent(logEvent); - int eventType = logEvent.getHeader().getType(); TableMeta tableMeta = null; boolean needDmlParse = false; @@ -347,7 +338,6 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement // clear for gc event.setBuffer(null); - event.setBinlogFileName(null); event.setEvent(null); event.setTable(null); event.setEntry(null); @@ -371,21 +361,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement class MessageEvent { - private String binlogFileName; // for local binlog parse private LogBuffer buffer; private CanalEntry.Entry entry; private boolean needDmlParse = false; private TableMeta table; private LogEvent event; - public String getBinlogFileName() { - return binlogFileName; - } - - public void setBinlogFileName(String binlogFileName) { - this.binlogFileName = binlogFileName; - } - public LogBuffer getBuffer() { return buffer; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index f5ff93ab..e161e34e 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -10,7 +10,6 @@ import java.util.Arrays; import java.util.BitSet; import java.util.List; -import com.taobao.tddl.dbsync.binlog.event.*; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; @@ -41,7 +40,23 @@ import com.alibaba.otter.canal.protocol.CanalEntry.Type; import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.google.protobuf.ByteString; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent; +import com.taobao.tddl.dbsync.binlog.event.HeartbeatLogEvent; +import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent; +import com.taobao.tddl.dbsync.binlog.event.LogHeader; +import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; +import com.taobao.tddl.dbsync.binlog.event.RandLogEvent; +import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer; +import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent; +import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo; +import com.taobao.tddl.dbsync.binlog.event.UnknownLogEvent; +import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.UserVarLogEvent; +import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.XidLogEvent; import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent; /** @@ -872,11 +887,14 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private TableMeta getTableMeta(String dbName, String tbName, boolean useCache, EntryPosition position) { try { return tableMetaCache.getTableMeta(dbName, tbName, useCache, position); - } catch (Exception e) { + } catch (Throwable e) { String message = ExceptionUtils.getRootCauseMessage(e); if (filterTableError) { if (StringUtils.contains(message, "errorNumber=1146") && StringUtils.contains(message, "doesn't exist")) { return null; + } else if (StringUtils.contains(message, "errorNumber=1142") + && StringUtils.contains(message, "command denied")) { + return null; } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java index 224852cc..eb227303 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java @@ -25,13 +25,14 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException; */ public class BinLogFileQueue { - private String baseName = "mysql-bin."; - private List binlogs = new ArrayList(); - private File directory; - private ReentrantLock lock = new ReentrantLock(); - private Condition nextCondition = lock.newCondition(); - private Timer timer = new Timer(true); - private long reloadInterval = 10 * 1000L; // 10秒 + private String baseName = "mysql-bin."; + private List binlogs = new ArrayList(); + private File directory; + private ReentrantLock lock = new ReentrantLock(); + private Condition nextCondition = lock.newCondition(); + private Timer timer = new Timer(true); + private long reloadInterval = 10 * 1000L; // 10秒 + private CanalParseException exception = null; public BinLogFileQueue(String directory){ this(new File(directory)); @@ -52,9 +53,24 @@ public class BinLogFileQueue { timer.scheduleAtFixedRate(new TimerTask() { public void run() { - List files = listBinlogFiles(); - for (File file : files) { - offer(file); + try { + // File errorFile = new File(BinLogFileQueue.this.directory, + // errorFileName); + // if (errorFile.isFile() && errorFile.exists()) { + // String text = StringUtils.join(IOUtils.readLines(new + // FileInputStream(errorFile)), "\n"); + // exception = new CanalParseException(text); + // } + List files = listBinlogFiles(); + for (File file : files) { + offer(file); + } + } catch (Throwable e) { + exception = new CanalParseException(e); + } + + if (exception != null) { + offer(null); } } }, reloadInterval, reloadInterval); @@ -69,6 +85,10 @@ public class BinLogFileQueue { public File getNextFile(File pre) { try { lock.lockInterruptibly(); + if (exception != null) { + throw exception; + } + if (binlogs.size() == 0) { return null; } else { @@ -94,6 +114,10 @@ public class BinLogFileQueue { public File getBefore(File file) { try { lock.lockInterruptibly(); + if (exception != null) { + throw exception; + } + if (binlogs.size() == 0) { return null; } else { @@ -130,6 +154,9 @@ public class BinLogFileQueue { nextCondition.await();// 等待新文件 } + if (exception != null) { + throw exception; + } if (pre == null) {// 第一次 return binlogs.get(0); } else { @@ -170,13 +197,16 @@ public class BinLogFileQueue { private boolean offer(File file) { try { lock.lockInterruptibly(); - if (!binlogs.contains(file)) { - binlogs.add(file); - nextCondition.signalAll();// 唤醒 - return true; - } else { - return false; + if (file != null) { + if (!binlogs.contains(file)) { + binlogs.add(file); + nextCondition.signalAll();// 唤醒 + return true; + } } + + nextCondition.signalAll();// 唤醒 + return false; } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java index d1716e18..e132514b 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java @@ -7,6 +7,8 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; import java.util.Collections; import java.util.Comparator; import java.util.Iterator; @@ -14,6 +16,9 @@ import java.util.LinkedList; import java.util.List; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; + +import javax.net.ssl.SSLContext; import org.apache.commons.compress.archivers.tar.TarArchiveEntry; import org.apache.commons.compress.archivers.tar.TarArchiveInputStream; @@ -23,8 +28,16 @@ import org.apache.commons.lang.StringUtils; import org.apache.http.HttpResponse; import org.apache.http.client.config.RequestConfig; import org.apache.http.client.methods.HttpGet; +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.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.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,7 +55,7 @@ public class BinlogDownloadQueue { private LinkedBlockingQueue downloadQueue = new LinkedBlockingQueue(); private LinkedBlockingQueue taskQueue = new LinkedBlockingQueue(); private LinkedList binlogList; - private final int batchSize; + private final int batchFileSize; private Thread downloadThread; public boolean running = true; private final String destDir; @@ -50,9 +63,9 @@ public class BinlogDownloadQueue { private int currentSize; private String lastDownload; - public BinlogDownloadQueue(List downloadQueue, int batchSize, String destDir) throws IOException{ + public BinlogDownloadQueue(List downloadQueue, int batchFileSize, String destDir) throws IOException{ this.binlogList = new LinkedList(downloadQueue); - this.batchSize = batchSize; + this.batchFileSize = batchFileSize; this.destDir = destDir; this.currentSize = 0; prepareBinlogList(); @@ -83,11 +96,12 @@ public class BinlogDownloadQueue { if (downloadThread != null) { return; } - downloadThread = new Thread(new DownloadThread()); + downloadThread = new Thread(new DownloadThread(), "download-" + destDir); + downloadThread.setDaemon(true); downloadThread.start(); } - public BinlogFile tryOne() throws IOException { + public BinlogFile tryOne() throws Throwable { BinlogFile binlogFile = binlogList.poll(); download(binlogFile); hostId = binlogFile.getHostInstanceID(); @@ -121,7 +135,7 @@ public class BinlogDownloadQueue { } public void prepare() throws InterruptedException { - for (int i = this.currentSize; i < batchSize && !binlogList.isEmpty(); i++) { + for (int i = this.currentSize; i < batchFileSize && !binlogList.isEmpty(); i++) { BinlogFile binlogFile = null; while (!binlogList.isEmpty()) { binlogFile = binlogList.poll(); @@ -150,11 +164,37 @@ public class BinlogDownloadQueue { downloadQueue.clear(); } - private void download(BinlogFile binlogFile) throws IOException { + private void download(BinlogFile binlogFile) throws Throwable { String downloadLink = binlogFile.getDownloadLink(); String fileName = binlogFile.getFileName(); + + downloadLink = downloadLink.trim(); + CloseableHttpClient httpClient = null; + if (downloadLink.startsWith("https")) { + HttpClientBuilder builder = HttpClientBuilder.create(); + builder.setMaxConnPerRoute(50); + builder.setMaxConnTotal(100); + // 创建支持忽略证书的https + final 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(); + } else { + httpClient = HttpClientBuilder.create().setMaxConnPerRoute(50).setMaxConnTotal(100).build(); + } + HttpGet httpGet = new HttpGet(downloadLink); - CloseableHttpClient httpClient = HttpClientBuilder.create().setMaxConnPerRoute(50).setMaxConnTotal(100).build(); RequestConfig requestConfig = RequestConfig.custom() .setConnectTimeout(TIMEOUT) .setConnectionRequestTimeout(TIMEOUT) @@ -187,6 +227,9 @@ public class BinlogDownloadQueue { String name = tarArchiveEntry.getName(); File tarFile = new File(parentFile, name + ".tmp"); logger.info("start to download file " + tarFile.getName()); + if (tarFile.exists()) { + tarFile.delete(); + } BufferedOutputStream bos = null; try { bos = new BufferedOutputStream(new FileOutputStream(tarFile)); @@ -204,6 +247,10 @@ public class BinlogDownloadQueue { tais.close(); } else { File file = new File(parentFile, fileName + ".tmp"); + if (file.exists()) { + file.delete(); + } + if (!file.isFile()) { file.createNewFile(); } @@ -245,17 +292,45 @@ public class BinlogDownloadQueue { @Override public void run() { while (running) { + BinlogFile binlogFile = null; try { - BinlogFile binlogFile = downloadQueue.poll(5000, TimeUnit.MILLISECONDS); + binlogFile = downloadQueue.poll(5000, TimeUnit.MILLISECONDS); if (binlogFile != null) { - download(binlogFile); + int retry = 1; + while (true) { + try { + download(binlogFile); + break; + } catch (Throwable e) { + if (retry % 10 == 0) { + retry = retry + 1; + try { + logger.warn("download failed + " + binlogFile.toString() + "], retry : " + + retry, e); + // File errorFile = new File(destDir, + // "error.txt"); + // FileWriter writer = new + // FileWriter(errorFile); + // writer.write(ExceptionUtils.getFullStackTrace(e)); + // writer.flush(); + // IOUtils.closeQuietly(writer); + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100 * retry)); + } catch (Throwable e1) { + logger.error("write error failed", e1); + } + } else { + retry = retry + 1; + } + } + } } + Runnable runnable = taskQueue.poll(5000, TimeUnit.MILLISECONDS); if (runnable != null) { runnable.run(); } - } catch (Exception e) { - e.printStackTrace(); + } catch (Throwable e) { + logger.error("task process failed", e); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java index 7b641ade..9ed865fe 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java @@ -155,6 +155,7 @@ public class HttpHelper { .register("https", new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE)) .build())) .build(); + // ---------------- 创建支持https 的client成功--------- URI uri = new URIBuilder(url).build(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java index 88caca53..3c38ed0c 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java @@ -4,93 +4,109 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; +import org.apache.commons.lang.StringUtils; + import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; import com.alibaba.otter.canal.parse.inbound.ParserExceptionHandler; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; /** + * aliyun rds的binlog parser支持 + * + * 
+ * 注意点：aliyun的binlog会有定期清理并备份到oss上, 这里实现了一份自动下载oss+rds binlog的机制
+ *
+ * * @author chengjin.lyf on 2018/7/20 上午10:52 * @since 1.0.25 */ public class RdsBinlogEventParserProxy extends MysqlEventParser { - private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址 - private String accesskey; // 云账号的ak - private String secretkey; // 云账号sk - private String instanceId; // rds实例id - private Long startTime; - private Long endTime; - private String directory; // binlog - // 目录 - private int batchSize = 4; // 最多下载的binlog文件数量 + private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址 + private String accesskey; // 云账号的ak + private String secretkey; // 云账号sk + private String instanceId; // rds实例id + private String directory; // binlog目录 + private int batchFileSize = 4; // 最多下载的binlog文件数量 - private RdsLocalBinlogEventParser rdsBinlogEventParser = new RdsLocalBinlogEventParser(); - private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() { + private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser(); + private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() { - @Override - public Thread newThread(Runnable r) { - Thread t = new Thread(r, "rds-binlog-daemon-thread"); - t.setDaemon(true); - return t; - } - }); + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r, + "rds-binlog-daemon-thread"); + t.setDaemon(true); + return t; + } + }); @Override public void start() { - final ParserExceptionHandler targetHandler = this.getParserExceptionHandler(); - rdsBinlogEventParser.setLogPositionManager(this.getLogPositionManager()); - rdsBinlogEventParser.setDestination(destination); - rdsBinlogEventParser.setAlarmHandler(this.getAlarmHandler()); - rdsBinlogEventParser.setConnectionCharset(this.connectionCharset); - rdsBinlogEventParser.setConnectionCharsetNumber(this.connectionCharsetNumber); - rdsBinlogEventParser.setEnableTsdb(this.enableTsdb); - rdsBinlogEventParser.setEventBlackFilter(this.eventBlackFilter); - rdsBinlogEventParser.setFilterQueryDcl(this.filterQueryDcl); - rdsBinlogEventParser.setFilterQueryDdl(this.filterQueryDdl); - rdsBinlogEventParser.setFilterQueryDml(this.filterQueryDml); - rdsBinlogEventParser.setFilterRows(this.filterRows); - rdsBinlogEventParser.setFilterTableError(this.filterTableError); - rdsBinlogEventParser.setIsGTIDMode(this.isGTIDMode); - rdsBinlogEventParser.setMasterInfo(this.masterInfo); - rdsBinlogEventParser.setEventFilter(this.eventFilter); - rdsBinlogEventParser.setMasterPosition(this.masterPosition); - rdsBinlogEventParser.setTransactionSize(this.transactionSize); - rdsBinlogEventParser.setUrl(this.rdsOpenApiUrl); - rdsBinlogEventParser.setAccesskey(this.accesskey); - rdsBinlogEventParser.setSecretkey(this.secretkey); - rdsBinlogEventParser.setInstanceId(this.instanceId); - rdsBinlogEventParser.setEventSink(eventSink); - rdsBinlogEventParser.setDirectory(directory); - rdsBinlogEventParser.setBatchSize(batchSize); - rdsBinlogEventParser.setFinishListener(new RdsLocalBinlogEventParser.ParseFinishListener() { - - @Override - public void onFinish() { - executorService.execute(new Runnable() { - - @Override - public void run() { - rdsBinlogEventParser.stop(); - RdsBinlogEventParserProxy.this.start(); - } - }); - + if (StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey) + && StringUtils.isNotEmpty(instanceId)) { + final ParserExceptionHandler targetHandler = this.getParserExceptionHandler(); + if (directory == null) { + directory = System.getProperty("java.io.tmpdir", "/tmp") + "/" + destination; } - }); - this.setParserExceptionHandler(new ParserExceptionHandler() { + rdsLocalBinlogEventParser.setLogPositionManager(this.getLogPositionManager()); + rdsLocalBinlogEventParser.setDestination(destination); + rdsLocalBinlogEventParser.setAlarmHandler(this.getAlarmHandler()); + rdsLocalBinlogEventParser.setConnectionCharset(this.connectionCharset); + rdsLocalBinlogEventParser.setConnectionCharsetNumber(this.connectionCharsetNumber); + rdsLocalBinlogEventParser.setEnableTsdb(this.enableTsdb); + rdsLocalBinlogEventParser.setEventBlackFilter(this.eventBlackFilter); + rdsLocalBinlogEventParser.setFilterQueryDcl(this.filterQueryDcl); + rdsLocalBinlogEventParser.setFilterQueryDdl(this.filterQueryDdl); + rdsLocalBinlogEventParser.setFilterQueryDml(this.filterQueryDml); + rdsLocalBinlogEventParser.setFilterRows(this.filterRows); + rdsLocalBinlogEventParser.setFilterTableError(this.filterTableError); + // rdsLocalBinlogEventParser.setIsGTIDMode(this.isGTIDMode); + rdsLocalBinlogEventParser.setMasterInfo(this.masterInfo); + rdsLocalBinlogEventParser.setEventFilter(this.eventFilter); + rdsLocalBinlogEventParser.setMasterPosition(this.masterPosition); + rdsLocalBinlogEventParser.setTransactionSize(this.transactionSize); + rdsLocalBinlogEventParser.setUrl(this.rdsOpenApiUrl); + rdsLocalBinlogEventParser.setAccesskey(this.accesskey); + rdsLocalBinlogEventParser.setSecretkey(this.secretkey); + rdsLocalBinlogEventParser.setInstanceId(this.instanceId); + rdsLocalBinlogEventParser.setEventSink(eventSink); + rdsLocalBinlogEventParser.setDirectory(directory); + rdsLocalBinlogEventParser.setBatchFileSize(batchFileSize); + rdsLocalBinlogEventParser.setParallel(this.parallel); + rdsLocalBinlogEventParser.setParallelBufferSize(this.parallelBufferSize); + rdsLocalBinlogEventParser.setParallelThreadSize(this.parallelThreadSize); + rdsLocalBinlogEventParser.setFinishListener(new RdsLocalBinlogEventParser.ParseFinishListener() { + + @Override + public void onFinish() { + executorService.execute(new Runnable() { + + @Override + public void run() { + rdsLocalBinlogEventParser.stop(); + RdsBinlogEventParserProxy.this.start(); + } + }); - @Override - public void handle(Throwable e) { - handleMysqlParserException(e); - if (targetHandler != null) { - targetHandler.handle(e); } - } - }); + }); + this.setParserExceptionHandler(new ParserExceptionHandler() { + + @Override + public void handle(Throwable e) { + handleMysqlParserException(e); + if (targetHandler != null) { + targetHandler.handle(e); + } + } + }); + } + super.start(); } - public void handleMysqlParserException(Throwable throwable) { + private void handleMysqlParserException(Throwable throwable) { if (throwable instanceof PositionNotFoundException) { logger.info("remove rds not found position, try download rds binlog!"); executorService.execute(new Runnable() { @@ -101,12 +117,12 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser { logger.info("stop mysql parser!"); RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this; long serverId = rdsBinlogEventParserProxy.getServerId(); - rdsBinlogEventParser.setServerId(serverId); + rdsLocalBinlogEventParser.setServerId(serverId); rdsBinlogEventParserProxy.stop(); logger.info("start rds mysql binlog parser!"); - rdsBinlogEventParser.start(); - } catch (Exception e) { - e.printStackTrace(); + rdsLocalBinlogEventParser.start(); + } catch (Throwable e) { + logger.info("handle exception failed", e); } } }); @@ -143,7 +159,8 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser { this.directory = directory; } - public void setBatchSize(int batchSize) { - this.batchSize = batchSize; + public void setBatchFileSize(int batchFileSize) { + this.batchFileSize = batchFileSize; } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java index 356a3520..2e79c536 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogOpenApi.java @@ -1,51 +1,20 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds; -import io.netty.handler.codec.http.HttpResponseStatus; - -import java.io.BufferedOutputStream; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.UnsupportedEncodingException; import java.net.URI; import java.net.URISyntaxException; -import java.net.URLEncoder; -import java.text.SimpleDateFormat; import java.util.Collections; import java.util.Date; -import java.util.HashMap; import java.util.List; -import java.util.Map; -import java.util.TimeZone; -import java.util.TreeMap; -import java.util.UUID; -import javax.crypto.Mac; -import javax.crypto.SecretKey; -import javax.crypto.spec.SecretKeySpec; - -import org.apache.commons.codec.binary.Base64; -import org.apache.commons.compress.archivers.tar.TarArchiveEntry; -import org.apache.commons.compress.archivers.tar.TarArchiveInputStream; -import org.apache.commons.io.FileUtils; -import org.apache.commons.io.IOUtils; import org.apache.commons.lang.StringUtils; -import org.apache.http.HttpResponse; -import org.apache.http.client.config.RequestConfig; -import org.apache.http.client.methods.HttpGet; -import org.apache.http.impl.client.CloseableHttpClient; -import org.apache.http.impl.client.HttpClientBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONArray; -import com.alibaba.fastjson.JSONObject; import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile; import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy; import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsItem; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.request.DescribeBackupPolicyRequest; import com.alibaba.otter.canal.parse.inbound.mysql.rds.request.DescribeBinlogFilesRequest; /** @@ -54,13 +23,7 @@ import com.alibaba.otter.canal.parse.inbound.mysql.rds.request.DescribeBinlogFil */ public class RdsBinlogOpenApi { - protected static final Logger logger = LoggerFactory.getLogger(RdsBinlogOpenApi.class); - private static final String ISO8601_DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss'Z'"; - private static final int TIMEOUT = 10000; - private static final String ENCODING = "UTF-8"; - private static final String MAC_NAME = "HmacSHA1"; - private static final String API_VERSION = "2014-08-15"; - private static final String SIGNATURE_VERSION = "1.0"; + protected static final Logger logger = LoggerFactory.getLogger(RdsBinlogOpenApi.class); public static List listBinlogFiles(String url, String ak, String sk, String dbInstanceId, Date startTime, Date endTime) { @@ -106,279 +69,32 @@ public class RdsBinlogOpenApi { return Collections.EMPTY_LIST; } - public static void downloadBinlogFiles(String url, String ak, String sk, String dbInstanceId, Date startTime, - Date endTime, File destDir) throws Throwable { - int pageSize = 100; - int pageNumber = 0; - int pageRecordCount = 1; - String hostInstanceID = null; - while (pageRecordCount > 0 && pageRecordCount <= pageSize) { - pageNumber += 1; - String result = describeBinlogFiles(url, ak, sk, dbInstanceId, startTime, endTime, pageSize, pageNumber); - JSONObject jsobObj = JSON.parseObject(result); - pageRecordCount = jsobObj.getInteger("PageRecordCount"); - - if (pageRecordCount > 0) { - FileUtils.forceMkdir(destDir); - File hostIdFile = new File(destDir, "hostId"); - if (hostIdFile.exists()) { - List lines = IOUtils.readLines(new FileInputStream(hostIdFile)); - hostInstanceID = StringUtils.join(lines, "\n"); - } - - String itemStr = jsobObj.getString("Items"); - JSONObject binLogFileObj = JSONObject.parseObject(itemStr); - JSONArray items = binLogFileObj.getJSONArray("BinLogFile"); - if (items == null || items.isEmpty()) { - continue; - } - for (int i = 0; i < items.size(); i++) { - JSONObject item = (JSONObject) items.get(i); - String oneHostInstanceID = item.getString("HostInstanceID"); - if (hostInstanceID == null) { - hostInstanceID = oneHostInstanceID; - FileOutputStream hostIdFileOut = null; - try { - hostIdFileOut = new FileOutputStream(hostIdFile); - hostIdFileOut.write(oneHostInstanceID.getBytes()); - hostIdFileOut.flush(); - } finally { - IOUtils.closeQuietly(hostIdFileOut); - } - } - - if (hostInstanceID.equals(oneHostInstanceID)) { // 只选择一个host下载 - String downloadLink = item.getString("DownloadLink"); - String fileName = StringUtils.substringBetween(downloadLink, "mysql-bin.", ".tar"); - if (StringUtils.isNotEmpty(fileName)) { - File currentFile = new File(destDir, "mysql-bin." + fileName); - if (currentFile.isFile() && currentFile.exists()) { - // 检查一下文件是否存在，存在就就没必要下载了 - continue; - } - } - - HttpGet httpGet = new HttpGet(downloadLink); - CloseableHttpClient httpClient = HttpClientBuilder.create() - .setMaxConnPerRoute(50) - .setMaxConnTotal(100) - .build(); - RequestConfig requestConfig = RequestConfig.custom() - .setConnectTimeout(TIMEOUT) - .setConnectionRequestTimeout(TIMEOUT) - .setSocketTimeout(TIMEOUT) - .build(); - httpGet.setConfig(requestConfig); - HttpResponse response = httpClient.execute(httpGet); - int statusCode = response.getStatusLine().getStatusCode(); - if (statusCode != HttpResponseStatus.OK.code()) { - throw new RuntimeException("download failed , url:" + downloadLink + " , statusCode:" - + statusCode); - } - saveFile(destDir, response); - } - } + public static RdsBackupPolicy queryBinlogBackupPolicy(String url, String ak, String sk, String dbInstanceId) { + DescribeBackupPolicyRequest request = new DescribeBackupPolicyRequest(); + if (StringUtils.isNotEmpty(url)) { + try { + URI uri = new URI(url); + request.setEndPoint(uri.getHost()); + } catch (URISyntaxException e) { + logger.error("resolve url host failed, will use default rds endpoint!"); } } - } - - private static void saveFile(File parentFile, HttpResponse response) throws IOException { - InputStream is = response.getEntity().getContent(); - long totalSize = Long.parseLong(response.getFirstHeader("Content-Length").getValue()); - String fileName = response.getFirstHeader("Content-Disposition").getValue(); - fileName = StringUtils.substringAfter(fileName, "filename="); - boolean isTar = StringUtils.endsWith(fileName, ".tar"); - FileUtils.forceMkdir(parentFile); - FileOutputStream fos = null; - try { - if (isTar) { - TarArchiveInputStream tais = new TarArchiveInputStream(is); - TarArchiveEntry tarArchiveEntry = null; - while ((tarArchiveEntry = tais.getNextTarEntry()) != null) { - String name = tarArchiveEntry.getName(); - File tarFile = new File(parentFile, name); - logger.info("start to download file " + tarFile.getName()); - BufferedOutputStream bos = null; - try { - bos = new BufferedOutputStream(new FileOutputStream(tarFile)); - int read = -1; - byte[] buffer = new byte[1024]; - while ((read = tais.read(buffer)) != -1) { - bos.write(buffer, 0, read); - } - logger.info("download file " + tarFile.getName() + " end!"); - } finally { - IOUtils.closeQuietly(bos); - } - } - tais.close(); - } else { - File file = new File(parentFile, fileName); - if (!file.isFile()) { - file.createNewFile(); + request.setRdsInstanceId(dbInstanceId); + request.setAccessKeyId(ak); + request.setAccessKeySecret(sk); + int retryTime = 3; + while (true) { + try { + return request.doAction(); + } catch (Exception e) { + if (retryTime-- <= 0) { + throw new RuntimeException(e); } try { - fos = new FileOutputStream(file); - byte[] buffer = new byte[1024]; - int len; - long copySize = 0; - long nextPrintProgress = 0; - logger.info("start to download file " + file.getName()); - while ((len = is.read(buffer)) != -1) { - fos.write(buffer, 0, len); - copySize += len; - long progress = copySize * 100 / totalSize; - if (progress >= nextPrintProgress) { - logger.info("download " + file.getName() + " progress : " + progress - + "% , download size : " + copySize + ", total size : " + totalSize); - nextPrintProgress += 10; - } - } - logger.info("download file " + file.getName() + " end!"); - fos.flush(); - } finally { - IOUtils.closeQuietly(fos); + Thread.sleep(100L); + } catch (InterruptedException e1) { } } - } finally { - IOUtils.closeQuietly(fos); } } - - public static String describeBinlogFiles(String url, String ak, String sk, String dbInstanceId, Date startTime, - Date endTime, int pageSize, int pageNumber) throws Exception { - Map paramMap = new HashMap(); - paramMap.put("Action", "DescribeBinlogFiles"); - paramMap.put("DBInstanceId", dbInstanceId); // rds实例id - paramMap.put("StartTime", formatIso8601Date(startTime)); - paramMap.put("EndTime", formatIso8601Date(endTime)); - paramMap.put("PageSize", String.valueOf(pageSize)); - paramMap.put("PageNumber", String.valueOf(pageNumber)); - return doRequest(url, paramMap, ak, sk); - } - - private static String doRequest(String domin, Map param, String ak, String sk) throws Exception { - param.put("AccessKeyId", ak); - param.put("SignatureMethod", "HMAC-SHA1"); - param.put("SignatureVersion", SIGNATURE_VERSION); - param.put("Version", API_VERSION); - param.put("SignatureNonce", UUID.randomUUID().toString()); - param.put("Format", "JSON"); - param.put("Timestamp", formatIso8601Date(new Date())); - String signStr = generate("POST", param, sk); - param.put("Signature", signStr); - String request = concatQueryString(param); - String url = domin + "?" + request; - String result = HttpHelper.post(url, null, Collections.EMPTY_MAP, TIMEOUT); - return result; - } - - public static String concatQueryString(Map parameters) throws UnsupportedEncodingException { - if (null == parameters) { - return null; - } - StringBuilder urlBuilder = new StringBuilder(""); - for (Map.Entry entry : parameters.entrySet()) { - String key = entry.getKey(); - String val = entry.getValue(); - urlBuilder.append(encode(key)); - if (val != null) { - urlBuilder.append("=").append(encode(val)); - } - urlBuilder.append("&"); - } - int strIndex = urlBuilder.length(); - if (parameters.size() > 0) { - urlBuilder.deleteCharAt(strIndex - 1); - } - return urlBuilder.toString(); - } - - public static String encode(String value) throws UnsupportedEncodingException { - return URLEncoder.encode(value, "UTF-8"); - } - - private static String formatIso8601Date(Date date) { - SimpleDateFormat df = new SimpleDateFormat(ISO8601_DATE_FORMAT); - df.setTimeZone(TimeZone.getTimeZone("GMT")); - return df.format(date); - } - - /** - * 使用 HMAC-SHA1 签名方法对对encryptText进行签名 - * - * @param encryptText 被签名的字符串 - * @param encryptKey 密钥 - * @return - * @throws Exception - */ - public static byte[] HmacSHA1Encrypt(String encryptText, String encryptKey) throws Exception { - byte[] data = encryptKey.getBytes(ENCODING); - // 根据给定的字节数组构造一个密钥,第二参数指定一个密钥算法的名称 - SecretKey secretKey = new SecretKeySpec(data, MAC_NAME); - // 生成一个指定 Mac 算法 的 Mac 对象 - Mac mac = Mac.getInstance(MAC_NAME); - // 用给定密钥初始化 Mac 对象 - mac.init(secretKey); - - byte[] text = encryptText.getBytes(ENCODING); - // 完成 Mac 操作 - return mac.doFinal(text); - } - - private static String base64(byte input[]) throws UnsupportedEncodingException { - return new String(Base64.encodeBase64(input), ENCODING); - } - - /** 对参数名称和参数值进行URL编码 **/ - public static String generate(String method, Map parameter, String accessKeySecret) - throws Exception { - String signString = generateSignString(method, parameter); - byte[] signBytes = HmacSHA1Encrypt(signString, accessKeySecret + "&"); - String signature = base64(signBytes); - if ("POST".equals(method)) { - return signature; - } - return URLEncoder.encode(signature, "UTF-8"); - } - - private static String generateQueryString(TreeMap treeMap) { - StringBuilder canonicalizedQueryString = new StringBuilder(); - boolean first = true; - for (String key : treeMap.navigableKeySet()) { - String value = treeMap.get(key); - if (!first) { - canonicalizedQueryString.append("&"); - } - first = false; - canonicalizedQueryString.append(percentEncode(key)).append("=").append(percentEncode(value)); - } - return canonicalizedQueryString.toString(); - } - - public static String generateSignString(String httpMethod, Map parameter) throws IOException { - TreeMap sortParameter = new TreeMap(); - sortParameter.putAll(parameter); - String canonicalizedQueryString = generateQueryString(sortParameter); - if (null == httpMethod) { - throw new RuntimeException("httpMethod can not be empty"); - } - /** 构造待签名的字符串* */ - StringBuilder stringToSign = new StringBuilder(); - stringToSign.append(httpMethod).append("&"); - stringToSign.append(percentEncode("/")).append("&"); - stringToSign.append(percentEncode(canonicalizedQueryString)); - return stringToSign.toString(); - } - - public static String percentEncode(String value) { - try { - return value == null ? null : URLEncoder.encode(value, ENCODING) - .replaceAll("\\+", "%20") - .replaceAll("\\*", "%2A") - .replaceAll("%7E", "~"); - } catch (Exception e) { - } - return ""; - } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java index 05a1935e..aac14d99 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java @@ -28,15 +28,15 @@ import com.alibaba.otter.canal.protocol.position.LogPosition; */ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements CanalEventParser, LocalBinLogConnection.FileParserListener { - private String url = "https://rds.aliyuncs.com/"; // openapi地址 - private String accesskey; // 云账号的ak - private String secretkey; // 云账号sk - private String instanceId; // rds实例id + private String url; // openapi地址 + private String accesskey; // 云账号的ak + private String secretkey; // 云账号sk + private String instanceId; // rds实例id private Long startTime; private Long endTime; private BinlogDownloadQueue binlogDownloadQueue; private ParseFinishListener finishListener; - private int batchSize; + private int batchFileSize; public RdsLocalBinlogEventParser(){ } @@ -65,10 +65,9 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements instanceId, new Date(startTime), new Date(endTime)); - binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchSize, directory); + binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchFileSize, directory); binlogDownloadQueue.silenceDownload(); needWait = true; - parallel = false; // try to download one file,use to test server id binlogDownloadQueue.tryOne(); } catch (Throwable e) { @@ -91,10 +90,12 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements binlogDownloadQueue.notifyNotMatch(); try { binlogDownloadQueue.cleanDir(); + binlogDownloadQueue.tryOne(); binlogDownloadQueue.prepare(); - } catch (Exception e) { + } catch (Throwable e) { throw new RuntimeException(e); } + try { binlogDownloadQueue.execute(new Runnable() { @@ -163,8 +164,10 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements } // 处理下logManager位点问题 LogPosition logPosition = logPositionManager.getLatestIndexBy(destination); - EntryPosition position = logPosition.getPostion(); - if (position != null) { + Long timestamp = 0L; + if (logPosition != null && logPosition.getPostion() != null) { + timestamp = logPosition.getPostion().getTimestamp(); + EntryPosition position = logPosition.getPostion(); LogPosition newLogPosition = new LogPosition(); String journalName = position.getJournalName(); int sepIdx = journalName.indexOf("."); @@ -181,9 +184,13 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements } if (binlogDownloadQueue.isLastFile(fileName)) { - logger.info("all file parse complete, switch to mysql parser!"); + logger.warn("last file : " + fileName + " , timestamp : " + timestamp + + " , all file parse complete, switch to mysql parser!"); finishListener.onFinish(); return; + } else { + logger.warn("parse local binlog file : " + fileName + " , timestamp : " + timestamp + + " , try the next binlog !"); } binlogDownloadQueue.prepare(); } catch (Exception e) { @@ -207,7 +214,7 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements void onFinish(); } - public void setBatchSize(int batchSize) { - this.batchSize = batchSize; + public void setBatchFileSize(int batchFileSize) { + this.batchFileSize = batchFileSize; } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java index e6b354d7..bee87521 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/BinlogFile.java @@ -6,7 +6,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.data; */ public class BinlogFile { - private Long FileSize; + private Long FileSize; private String LogBeginTime; private String LogEndTime; private String DownloadLink; @@ -69,4 +69,12 @@ public class BinlogFile { public void setFileName(String fileName) { this.fileName = fileName; } + + @Override + public String toString() { + return "BinlogFile [FileSize=" + FileSize + ", LogBeginTime=" + LogBeginTime + ", LogEndTime=" + LogEndTime + + ", DownloadLink=" + DownloadLink + ", HostInstanceID=" + HostInstanceID + ", LinkExpiredTime=" + + LinkExpiredTime + ", fileName=" + fileName + "]"; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java index d8d54e44..d5b6820d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/DescribeBinlogFileResult.java @@ -5,12 +5,13 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.data; * @since 1.0.25 */ public class DescribeBinlogFileResult { + private RdsItem Items; - private long PageNumber; - private long TotalRecordCount; - private long TotalFileSize; - private String RequestId; - private long PageRecordCount; + private long PageNumber; + private long TotalRecordCount; + private long TotalFileSize; + private String RequestId; + private long PageRecordCount; public RdsItem getItems() { return Items; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java index b6ca7c53..e007671a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsBackupPolicy.java @@ -9,15 +9,16 @@ public class RdsBackupPolicy { /** * 数据备份保留天数（7到730天）。 */ - private String BackupRetentionPeriod; + private String BackupRetentionPeriod; /** * 数据备份时间，格式：HH:mmZ- HH:mm Z。 */ - private String PreferredBackupTime; + private String PreferredBackupTime; /** - * 数据备份周期。Monday：周一；Tuesday：周二；Wednesday：周三；Thursday：周四；Friday：周五；Saturday：周六；Sunday：周日。 + * 数据备份周期。Monday：周一；Tuesday：周二；Wednesday：周三；Thursday：周四；Friday：周五；Saturday： + * 周六；Sunday：周日。 */ - private String PreferredBackupPeriod; + private String PreferredBackupPeriod; /** * 日志备份状态。Enable：开启；Disabled：关闭。 */ @@ -25,7 +26,7 @@ public class RdsBackupPolicy { /** * 日志备份保留天数（7到730天）。 */ - private int LogBackupRetentionPeriod; + private int LogBackupRetentionPeriod; public String getBackupRetentionPeriod() { return BackupRetentionPeriod; @@ -66,4 +67,12 @@ public class RdsBackupPolicy { public void setLogBackupRetentionPeriod(int logBackupRetentionPeriod) { LogBackupRetentionPeriod = logBackupRetentionPeriod; } + + @Override + public String toString() { + return "RdsBackupPolicy [BackupRetentionPeriod=" + BackupRetentionPeriod + ", PreferredBackupTime=" + + PreferredBackupTime + ", PreferredBackupPeriod=" + PreferredBackupPeriod + ", BackupLog=" + BackupLog + + ", LogBackupRetentionPeriod=" + LogBackupRetentionPeriod + "]"; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java index a2cfaffe..81c484e3 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/data/RdsItem.java @@ -7,6 +7,7 @@ import java.util.List; * @since 1.0.25 */ public class RdsItem { + private List BinLogFile; public List getBinLogFile() { @@ -16,4 +17,10 @@ public class RdsItem { public void setBinLogFile(List binLogFile) { BinLogFile = binLogFile; } + + @Override + public String toString() { + return "RdsItem [BinLogFile=" + BinLogFile + "]"; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java index 976968f6..f43523b5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java @@ -1,12 +1,18 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.request; +import io.netty.handler.codec.http.HttpResponseStatus; + import java.io.IOException; import java.io.UnsupportedEncodingException; import java.net.URLEncoder; import java.security.cert.CertificateException; import java.security.cert.X509Certificate; import java.text.SimpleDateFormat; -import java.util.*; +import java.util.Date; +import java.util.Map; +import java.util.TimeZone; +import java.util.TreeMap; +import java.util.UUID; import java.util.concurrent.TimeUnit; import javax.crypto.Mac; @@ -32,8 +38,6 @@ import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; import org.apache.http.ssl.SSLContexts; import org.apache.http.util.EntityUtils; -import io.netty.handler.codec.http.HttpResponseStatus; - /** * @author chengjin.lyf on 2018/8/7 下午2:26 * @since 1.0.25 @@ -49,42 +53,38 @@ public abstract class AbstractRequest { */ private static final String MAC_NAME = "HmacSHA1"; - private String accessKeyId; + private String accessKeyId; - private String accessKeySecret; + private String accessKeySecret; /** - * api 版本 - * + * api 版本 */ - private String version; + private String version; - private String endPoint = "rds.aliyuncs.com"; + private String endPoint = "rds.aliyuncs.com"; - private String protocol = "http"; + private String protocol = "http"; public void setProtocol(String protocol) { this.protocol = protocol; } - private int timeout = (int) TimeUnit.MINUTES.toMillis(1); - + private int timeout = (int) TimeUnit.MINUTES.toMillis(1); private Map treeMap = new TreeMap(); - public void putQueryString(String name, String value){ - if (StringUtils.isBlank(name) || StringUtils.isBlank(value)){ + public void putQueryString(String name, String value) { + if (StringUtils.isBlank(name) || StringUtils.isBlank(value)) { return; } treeMap.put(name, value); } - public void setVersion(String version) { this.version = version; } - public void setEndPoint(String endPoint) { this.endPoint = endPoint; } @@ -169,7 +169,7 @@ public abstract class AbstractRequest { p.put("Format", "JSON"); p.put("Version", version); p.put("AccessKeyId", accessKeyId); - p.put("SignatureMethod", "HMAC-SHA1"); //此处不能用变量 MAC_NAME + p.put("SignatureMethod", "HMAC-SHA1"); // 此处不能用变量 MAC_NAME p.put("Timestamp", formatUTCTZ(new Date())); p.put("SignatureVersion", "1.0"); p.put("SignatureNonce", UUID.randomUUID().toString()); @@ -193,6 +193,7 @@ public abstract class AbstractRequest { * @return * @throws IOException */ + @SuppressWarnings("deprecation") private final HttpResponse executeHttpRequest(HttpGet getMethod, String host) throws Exception { SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, new TrustStrategy() { @@ -202,24 +203,24 @@ public abstract class AbstractRequest { } }).build(); SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext, - new String[] { "TLSv1" }, - null, - SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER); + new String[] { "TLSv1" }, + null, + SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER); Registry registry = RegistryBuilder.create() - .register("http", PlainConnectionSocketFactory.INSTANCE) - .register("https", sslsf) - .build(); + .register("http", PlainConnectionSocketFactory.INSTANCE) + .register("https", sslsf) + .build(); HttpClientConnectionManager httpClientConnectionManager = new PoolingHttpClientConnectionManager(registry); CloseableHttpClient httpClient = HttpClientBuilder.create() - .setMaxConnPerRoute(50) - .setMaxConnTotal(100) - .setConnectionManager(httpClientConnectionManager) - .build(); + .setMaxConnPerRoute(50) + .setMaxConnTotal(100) + .setConnectionManager(httpClientConnectionManager) + .build(); RequestConfig requestConfig = RequestConfig.custom() - .setConnectTimeout(timeout) - .setConnectionRequestTimeout(timeout) - .setSocketTimeout(timeout) - .build(); + .setConnectTimeout(timeout) + .setConnectionRequestTimeout(timeout) + .setSocketTimeout(timeout) + .build(); getMethod.setConfig(requestConfig); HttpResponse response = httpClient.execute(getMethod); int statusCode = response.getStatusLine().getStatusCode(); @@ -232,14 +233,14 @@ public abstract class AbstractRequest { protected abstract T processResult(HttpResponse response) throws Exception; - protected void processBefore(){ + protected void processBefore() { } - public final T doAction() throws Exception { + public final T doAction() throws Exception { processBefore(); String requestStr = makeRequestString(treeMap); - HttpGet httpGet = new HttpGet(protocol + "://" +endPoint + "?" + requestStr); + HttpGet httpGet = new HttpGet(protocol + "://" + endPoint + "?" + requestStr); HttpResponse response = executeHttpRequest(httpGet, endPoint); if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) { String result = EntityUtils.toString(response.getEntity()); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java index 0fbdfc47..cf51961a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBackupPolicyRequest.java @@ -1,27 +1,26 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.request; -import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy; import org.apache.http.HttpResponse; import org.apache.http.util.EntityUtils; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy; /** * rds 备份策略查询 + * * @author chengjin.lyf on 2018/8/7 下午3:41 * @since 1.0.25 */ public class DescribeBackupPolicyRequest extends AbstractRequest { - - public DescribeBackupPolicyRequest() { + public DescribeBackupPolicyRequest(){ setVersion("2014-08-15"); putQueryString("Action", "DescribeBackupPolicy"); } - public void setRdsInstanceId(String rdsInstanceId) { putQueryString("DBInstanceId", rdsInstanceId); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java index 1c91c255..8aa90569 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/DescribeBinlogFilesRequest.java @@ -2,12 +2,12 @@ package com.alibaba.otter.canal.parse.inbound.mysql.rds.request; import java.util.Date; -import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult; import org.apache.http.HttpResponse; import org.apache.http.util.EntityUtils; import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.TypeReference; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.DescribeBinlogFileResult; /** * @author chengjin.lyf on 2018/8/7 下午3:41 @@ -15,11 +15,9 @@ import com.alibaba.fastjson.TypeReference; */ public class DescribeBinlogFilesRequest extends AbstractRequest { - - public DescribeBinlogFilesRequest() { + public DescribeBinlogFilesRequest(){ setVersion("2014-08-15"); putQueryString("Action", "DescribeBinlogFiles"); - } public void setRdsInstanceId(String rdsInstanceId) { @@ -35,11 +33,11 @@ public class DescribeBinlogFilesRequest extends AbstractRequest() { - }); + DescribeBinlogFileResult describeBinlogFileResult = JSONObject.parseObject(result, + new TypeReference() { + }); return describeBinlogFileResult; } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java index 9a886336..904b55a9 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java @@ -68,7 +68,8 @@ public class MemoryTableMeta implements TableMetaTSDB { try { // druid暂时flush privileges语法解析有问题 - if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush")) { + if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush") + && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "grant")) { repository.console(ddl); } } catch (Throwable e) { diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java index 986d4d6a..546ee58c 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogEventParserProxyTest.java @@ -8,6 +8,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.junit.Assert; import org.junit.Test; +import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; import com.alibaba.otter.canal.parse.helper.TimeoutChecker; import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogEventParserProxy; import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; @@ -28,10 +29,9 @@ public class RdsBinlogEventParserProxyTest { private static final String MYSQL_ADDRESS = ""; private static final String USERNAME = ""; private static final String PASSWORD = ""; - public static final String DBNAME = ""; - public static final String TBNAME = ""; - public static final String DDL = ""; - + public static final String DBNAME = ""; + public static final String TBNAME = ""; + public static final String DDL = ""; @Test public void test_timestamp() throws InterruptedException { @@ -41,7 +41,7 @@ public class RdsBinlogEventParserProxyTest { final RdsBinlogEventParserProxy controller = new RdsBinlogEventParserProxy(); Calendar calendar = Calendar.getInstance(); - calendar.add(Calendar.DAY_OF_YEAR, -1); + calendar.add(Calendar.HOUR_OF_DAY, -24 * 4); final EntryPosition defaultPosition = buildPosition(null, null, calendar.getTimeInMillis()); controller.setSlaveId(3344L); controller.setDetectingEnable(false); @@ -51,13 +51,15 @@ public class RdsBinlogEventParserProxyTest { controller.setInstanceId(""); controller.setAccesskey(""); controller.setSecretkey(""); - controller.setBatchSize(4); -// controller.setRdsOpenApiUrl("https://rds.aliyuncs.com/"); + controller.setDirectory("/tmp/binlog"); + controller.setEventBlackFilter(new AviaterRegexFilter("mysql\\.*")); + controller.setFilterTableError(true); + controller.setBatchFileSize(4); controller.setEventSink(new AbstractCanalEventSinkTest>() { @Override public boolean sink(List entrys, InetSocketAddress remoteAddress, String destination) - throws CanalSinkException { + throws CanalSinkException { for (CanalEntry.Entry entry : entrys) { if (entry.getEntryType() != CanalEntry.EntryType.HEARTBEAT) { entryCount.incrementAndGet(); @@ -79,8 +81,8 @@ public class RdsBinlogEventParserProxyTest { controller.setLogPositionManager(new AbstractLogPositionManager() { private LogPosition logPosition; + public void persistLogPosition(String destination, LogPosition logPosition) { - System.out.println(logPosition); this.logPosition = logPosition; } @@ -105,7 +107,6 @@ public class RdsBinlogEventParserProxyTest { Assert.assertTrue(entryPosition.getTimestamp() <= defaultPosition.getTimestamp()); } - // ======================== helper method ======================= private EntryPosition buildPosition(String binlogFile, Long offest, Long timestamp) { diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java index 596847ae..25a14284 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/RdsBinlogOpenApiTest.java @@ -1,11 +1,14 @@ package com.alibaba.otter.canal.parse.inbound.mysql; -import java.io.File; import java.util.Date; +import java.util.List; import org.apache.commons.lang.time.DateUtils; +import org.junit.Test; import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogOpenApi; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile; +import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.RdsBackupPolicy; /** * @author agapple 2017年10月15日 下午2:14:34 @@ -13,15 +16,19 @@ import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogOpenApi; */ public class RdsBinlogOpenApiTest { + @Test public void testSimple() throws Throwable { - Date startTime = DateUtils.parseDate("2017-10-13 20:56:58", new String[] { "yyyy-MM-dd HH:mm:ss" }); - Date endTime = DateUtils.parseDate("2017-10-14 02:57:59", new String[] { "yyyy-MM-dd HH:mm:ss" }); - RdsBinlogOpenApi.downloadBinlogFiles("https://rds.aliyuncs.com/", - "", - "", - "rm-bp180v4mfjnm157es", - startTime, - endTime, - new File("/tmp/binlog/")); + Date startTime = DateUtils.parseDate("2018-08-10 12:00:00", new String[] { "yyyy-MM-dd HH:mm:ss" }); + Date endTime = DateUtils.parseDate("2018-08-11 12:00:00", new String[] { "yyyy-MM-dd HH:mm:ss" }); + String url = "https://rds.aliyuncs.com/"; + String ak = ""; + String sk = ""; + String dbInstanceId = ""; + + RdsBackupPolicy backupPolicy = RdsBinlogOpenApi.queryBinlogBackupPolicy(url, ak, sk, dbInstanceId); + System.out.println(backupPolicy); + + List binlogFiles = RdsBinlogOpenApi.listBinlogFiles(url, ak, sk, dbInstanceId, startTime, endTime); + System.out.println(binlogFiles); } } From 11d4761bcb89077c6af358d95aa82d870cf069af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Tue, 14 Aug 2018 13:23:24 +0800 Subject: [PATCH 2/9] fixed docker build --- .../otter/canal/deployer/CanalConstants.java | 1 - .../otter/canal/deployer/CanalController.java | 8 ---- deployer/src/main/resources/canal.properties | 1 - docker/Dockerfile | 35 +--------------- docker/base/Dockerfile | 41 +++++++++++++++++++ docker/build.sh | 22 +++++----- docker/image/admin/app.sh | 4 +- docker/image/admin/health.sh | 9 +++- .../src/test/resources/canal.properties | 2 +- .../canal/prometheus/PrometheusService.java | 12 ++---- .../prometheus/impl/ParserCollector.java | 1 + .../canal/prometheus/impl/StoreCollector.java | 6 +-- .../embedded/CanalServerWithEmbedded.java | 6 --- .../otter/canal/spi/CanalMetricsService.java | 5 --- .../canal/spi/NopCanalMetricsService.java | 5 --- .../memory/MemoryEventStoreWithBuffer.java | 13 +++--- 16 files changed, 74 insertions(+), 97 deletions(-) create mode 100644 docker/base/Dockerfile 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 c7a90d30..b328feb7 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 @@ -15,7 +15,6 @@ public class CanalConstants { public static final String CANAL_ID = ROOT + "." + "id"; public static final String CANAL_IP = ROOT + "." + "ip"; public static final String CANAL_PORT = ROOT + "." + "port"; - public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port"; public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers"; public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty"; 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 53a2b9b5..6b0fc934 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 @@ -98,14 +98,6 @@ public class CanalController { port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT)); embededCanalServer = CanalServerWithEmbedded.instance(); embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator - try { - int metricsPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_METRICS_PULL_PORT)); - embededCanalServer.setMetricsPort(metricsPort); - } catch (NumberFormatException e) { - logger.info("No valid metrics server port found, use default 11112."); - embededCanalServer.setMetricsPort(11112); - } - String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY); if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) { canalServer = CanalServerWithNetty.instance(); diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 1cefb7d9..9083477d 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -4,7 +4,6 @@ canal.id= 1 canal.ip= canal.port=11111 -canal.metrics.pull.port=11112 canal.zkServers= # flush data to zk canal.zookeeper.flush.period = 1000 diff --git a/docker/Dockerfile b/docker/Dockerfile index 2867b955..835e881a 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -1,32 +1,7 @@ -FROM centos:centos6.7 +FROM canal/osbase:v1 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/ @@ -39,14 +14,6 @@ RUN \ 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 && \ diff --git a/docker/base/Dockerfile b/docker/base/Dockerfile new file mode 100644 index 00000000..010b97e0 --- /dev/null +++ b/docker/base/Dockerfile @@ -0,0 +1,41 @@ +FROM centos:centos6.10 + +MAINTAINER agapple (jianghang115@gmail.com) + +env DOWNLOAD_LINK="http://download.oracle.com/otn-pub/java/jdk/8u181-b13/96a7b8442fe848ef90c96a2fad6ed6d1/jdk-8u181-linux-x64.rpm" +# 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 && \ + true + +RUN \ + touch /var/lib/rpm/* && \ + wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=xxx; oraclelicense=accept-securebackup-cookie" "$DOWNLOAD_LINK" -O /tmp/jdk-8-linux-x64.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 && \ + yum clean all && \ + true + +CMD ["/bin/bash"] diff --git a/docker/build.sh b/docker/build.sh index 58424fb3..eab5b2b3 100644 --- a/docker/build.sh +++ b/docker/build.sh @@ -14,17 +14,15 @@ case "`uname`" in 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/ +if [ "$1" == "base" ] ; then + docker build --no-cache -t canal/osbase $BASE/base else - cp $BASE/../target/canal.deployer-*.tar.gz $BASE/ - docker build --no-cache -t canal/canal-server $BASE/ + 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 fi diff --git a/docker/image/admin/app.sh b/docker/image/admin/app.sh index 254f7e48..66928ac3 100755 --- a/docker/image/admin/app.sh +++ b/docker/image/admin/app.sh @@ -86,7 +86,9 @@ function start_canal() { echo "multi destination:$destination is not support" exit 1; else - mv /home/admin/canal-server/conf/example /home/admin/canal-server/conf/$destination + if [ "$destination" != "" ] && [ "$destination" != "example" ] ; then + mv /home/admin/canal-server/conf/example /home/admin/canal-server/conf/$destination + fi fi su admin -c 'cd /home/admin/canal-server/bin/ && sh restart.sh 1>>/tmp/start.log 2>&1' sleep 5 diff --git a/docker/image/admin/health.sh b/docker/image/admin/health.sh index 17a89740..c48e6d9d 100755 --- a/docker/image/admin/health.sh +++ b/docker/image/admin/health.sh @@ -1,6 +1,11 @@ #!/bin/sh -CHECK_URL="http://127.0.0.1:8080/metrics" -CHECK_POINT="success" +metrics_port=`perl -le 'print $ENV{"canal.metrics.pull.port"}'` +if [ "$metrics_port" == "" ]; then + metrics_port="11112" +fi + +CHECK_URL="http://127.0.0.1:$metrics_port/metrics" +CHECK_POINT="canal" CHECK_COUNT=`curl -s --connect-timeout 7 --max-time 7 $CHECK_URL | grep -c $CHECK_POINT` if [ $CHECK_COUNT -eq 0 ]; then echo "[FAILED]" diff --git a/instance/spring/src/test/resources/canal.properties b/instance/spring/src/test/resources/canal.properties index ffb5cf0e..1887781d 100644 --- a/instance/spring/src/test/resources/canal.properties +++ b/instance/spring/src/test/resources/canal.properties @@ -51,4 +51,4 @@ canal.instance.global.mode = spring canal.instance.global.lazy = false #canal.instance.global.manager.address = 127.0.0.1:1099 canal.instance.global.spring.xml = classpath:spring/memory-instance.xml -#canal.instance.global.spring.xml = classpath:spring/default-instance.xml +#canal.instance.global.spring.xml = classpath:spring/default-instance.xml \ No newline at end of file diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java index 18bf63f7..485f6728 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java @@ -22,7 +22,6 @@ public class PrometheusService implements CanalMetricsService { private static final Logger logger = LoggerFactory.getLogger(PrometheusService.class); private final CanalInstanceExports instanceExports; private volatile boolean running = false; - private int port; private HTTPServer server; private final ClientInstanceProfiler clientProfiler; @@ -42,9 +41,9 @@ public class PrometheusService implements CanalMetricsService { @Override public void initialize() { try { - logger.info("Start prometheus HTTPServer on port {}.", port); - //TODO 2.Https? - server = new HTTPServer(port); + //TODO 1.Configurable port + //TODO 2.Https + server = new HTTPServer(11112); } catch (IOException e) { logger.warn("Unable to start prometheus HTTPServer.", e); return; @@ -113,9 +112,4 @@ public class PrometheusService implements CanalMetricsService { logger.info("Unregister metrics for destination {}.", instance.getDestination()); } - @Override - public void setServerPort(int port) { - this.port = port; - } - } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java index c1883658..73102795 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java @@ -21,6 +21,7 @@ import java.util.concurrent.atomic.AtomicLong; import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST; import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; +import static java.util.concurrent.TimeUnit.NANOSECONDS; /** * @author Chuanyi Li diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java index d345602c..3be87c3e 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java @@ -40,7 +40,7 @@ public class StoreCollector extends Collector implements InstanceRegistry { private static final String PRODUCE_MEM_HELP = "Produced mem bytes of canal instance"; private static final String CONSUME_MEM_HELP = "Consumed mem bytes of canal instance"; private final ConcurrentMap instances = new ConcurrentHashMap(); - private final List storeLabelsList = Arrays.asList(DEST, "batchMode", "size"); + private final List storeLabelsList = Arrays.asList(DEST, "batchMode"); private StoreCollector() {} @@ -100,8 +100,7 @@ public class StoreCollector extends Collector implements InstanceRegistry { holder.putSeq = memStore.getPutSequence(); holder.ackSeq = memStore.getAckSequence(); holder.destLabelValues = Collections.singletonList(destination); - holder.size = memStore.getBufferSize(); - holder.storeLabelValues = Arrays.asList(destination, holder.batchMode.name(), Integer.toString(holder.size)); + holder.storeLabelValues = Arrays.asList(destination, memStore.getBatchMode().name()); Preconditions.checkNotNull(holder.batchMode); Preconditions.checkNotNull(holder.putSeq); Preconditions.checkNotNull(holder.ackSeq); @@ -129,7 +128,6 @@ public class StoreCollector extends Collector implements InstanceRegistry { private BatchMode batchMode; private AtomicLong putMemSize; private AtomicLong ackMemSize; - private int size; private List destLabelValues; private List storeLabelValues; } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index 9ac5a834..ccf21020 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -44,7 +44,6 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C private Map canalInstances; // private Map lastRollbackPostions; private CanalInstanceGenerator canalInstanceGenerator; - private int metricsPort; private CanalMetricsService metrics = NopCanalMetricsService.NOP; private static class SingletonHolder { @@ -65,7 +64,6 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C super.start(); // 如果存在provider,则启动metrics service loadCanalMetrics(); - metrics.setServerPort(metricsPort); metrics.initialize(); canalInstances = MigrateMap.makeComputingMap(new Function() { @@ -531,8 +529,4 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C this.canalInstanceGenerator = canalInstanceGenerator; } - public void setMetricsPort(int metricsPort) { - this.metricsPort = metricsPort; - } - } diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java index a4528800..6d823be8 100644 --- a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java +++ b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java @@ -39,9 +39,4 @@ public interface CanalMetricsService { */ void unregister(CanalInstance instance); - /** - * @param port server port for pull - */ - void setServerPort(int port); - } diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java b/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java index 45926348..0d6ad088 100644 --- a/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java +++ b/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java @@ -35,9 +35,4 @@ public class NopCanalMetricsService implements CanalMetricsService { public void unregister(CanalInstance instance) { } - - @Override - public void setServerPort(int port) { - - } } diff --git a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java index ef97e206..00ba2e71 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java @@ -545,9 +545,6 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple } // ================ setter / getter ================== - public int getBufferSize() { - return this.bufferSize; - } public void setBufferSize(int bufferSize) { this.bufferSize = bufferSize; @@ -566,22 +563,22 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple } public AtomicLong getPutSequence() { - return this.putSequence; + return putSequence; } public AtomicLong getAckSequence() { - return this.ackSequence; + return ackSequence; } public AtomicLong getPutMemSize() { - return this.putMemSize; + return putMemSize; } public AtomicLong getAckMemSize() { - return this.ackMemSize; + return ackMemSize; } public BatchMode getBatchMode() { - return this.batchMode; + return batchMode; } } From 8a74504fe3ec5c149b896734d052bb5798b73462 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Tue, 14 Aug 2018 13:29:03 +0800 Subject: [PATCH 3/9] preapre for 1.1.0 release --- client/pom.xml | 2 +- common/pom.xml | 2 +- dbsync/pom.xml | 2 +- deployer/pom.xml | 2 +- docker/Dockerfile | 1 - driver/pom.xml | 2 +- example/pom.xml | 2 +- filter/pom.xml | 2 +- instance/core/pom.xml | 2 +- instance/manager/pom.xml | 2 +- instance/pom.xml | 2 +- instance/spring/pom.xml | 2 +- kafka-client/pom.xml | 4 ++-- kafka/pom.xml | 4 ++-- meta/pom.xml | 2 +- parse/pom.xml | 2 +- pom.xml | 2 +- prometheus/pom.xml | 4 ++-- protocol/pom.xml | 2 +- server/pom.xml | 2 +- sink/pom.xml | 2 +- store/pom.xml | 2 +- 22 files changed, 24 insertions(+), 25 deletions(-) diff --git a/client/pom.xml b/client/pom.xml index c736b141..038f2a9f 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/common/pom.xml b/common/pom.xml index 9c57c560..afa3372c 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml canal.common diff --git a/dbsync/pom.xml b/dbsync/pom.xml index 1f274e82..17ec6c30 100644 --- a/dbsync/pom.xml +++ b/dbsync/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/deployer/pom.xml b/deployer/pom.xml index 68224ca1..2de1ce1e 100644 --- a/deployer/pom.xml +++ b/deployer/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/docker/Dockerfile b/docker/Dockerfile index 835e881a..311ac0b5 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -5,7 +5,6 @@ MAINTAINER agapple (jianghang115@gmail.com) # 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 && \ diff --git a/driver/pom.xml b/driver/pom.xml index edd24e57..648e64cc 100644 --- a/driver/pom.xml +++ b/driver/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/example/pom.xml b/example/pom.xml index 2897e821..4f13d017 100644 --- a/example/pom.xml +++ b/example/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/filter/pom.xml b/filter/pom.xml index 30b53e28..f662f5ea 100644 --- a/filter/pom.xml +++ b/filter/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/instance/core/pom.xml b/instance/core/pom.xml index cbfab2fd..e3812d4e 100644 --- a/instance/core/pom.xml +++ b/instance/core/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../../pom.xml canal.instance.core diff --git a/instance/manager/pom.xml b/instance/manager/pom.xml index 10f01e43..79340bd0 100644 --- a/instance/manager/pom.xml +++ b/instance/manager/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../../pom.xml com.alibaba.otter diff --git a/instance/pom.xml b/instance/pom.xml index aeec17cb..f45baee1 100644 --- a/instance/pom.xml +++ b/instance/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/instance/spring/pom.xml b/instance/spring/pom.xml index 6a8f172a..116dee77 100644 --- a/instance/spring/pom.xml +++ b/instance/spring/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../../pom.xml com.alibaba.otter diff --git a/kafka-client/pom.xml b/kafka-client/pom.xml index d814be52..5444af5d 100644 --- a/kafka-client/pom.xml +++ b/kafka-client/pom.xml @@ -6,7 +6,7 @@ canal com.alibaba.otter - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter @@ -108,4 +108,4 @@ - \ No newline at end of file + diff --git a/kafka/pom.xml b/kafka/pom.xml index 8d3b35d7..d46a2d03 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -6,7 +6,7 @@ canal com.alibaba.otter - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter @@ -137,4 +137,4 @@ - \ No newline at end of file + diff --git a/meta/pom.xml b/meta/pom.xml index 6192d92d..93c4ad7b 100644 --- a/meta/pom.xml +++ b/meta/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/parse/pom.xml b/parse/pom.xml index 994d06f0..fb93fc99 100644 --- a/parse/pom.xml +++ b/parse/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml canal.parse diff --git a/pom.xml b/pom.xml index 56ac821a..3101aca6 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ canal pom canal module for otter ${project.version} - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT https://github.com/alibaba/canal org.sonatype.oss diff --git a/prometheus/pom.xml b/prometheus/pom.xml index b814c7e6..7d192834 100644 --- a/prometheus/pom.xml +++ b/prometheus/pom.xml @@ -5,13 +5,13 @@ canal com.alibaba.otter - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT 4.0.0 com.alibaba.otter canal.prometheus - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT canal prometheus module for otter ${project.version} diff --git a/protocol/pom.xml b/protocol/pom.xml index ec7c7867..94715d15 100644 --- a/protocol/pom.xml +++ b/protocol/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/server/pom.xml b/server/pom.xml index 1d243981..b7f83d56 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml canal.server diff --git a/sink/pom.xml b/sink/pom.xml index 4ed7e04f..9f136d35 100644 --- a/sink/pom.xml +++ b/sink/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/store/pom.xml b/store/pom.xml index 318c2f9d..9dee6737 100644 --- a/store/pom.xml +++ b/store/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.0.26-SNAPSHOT + 1.1.0-SNAPSHOT ../pom.xml com.alibaba.otter From 13291c8d7bd76c71cfdf3e1a058409393ee16e18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Tue, 14 Aug 2018 13:38:01 +0800 Subject: [PATCH 4/9] fixed docker build --- docker/build.sh | 1 + docker/run.sh | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/docker/build.sh b/docker/build.sh index eab5b2b3..6b42e3a4 100644 --- a/docker/build.sh +++ b/docker/build.sh @@ -17,6 +17,7 @@ BASE=${bin_abs_path} if [ "$1" == "base" ] ; then docker build --no-cache -t canal/osbase $BASE/base else + rm -rf $BASE/canal.*.tar.gz ; 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/ diff --git a/docker/run.sh b/docker/run.sh index 7c56e2ba..3d5f7659 100644 --- a/docker/run.sh +++ b/docker/run.sh @@ -10,7 +10,7 @@ function usage() { 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=.*\\..* " + echo " -e canal.instance.filter.regex=.*\\\\\\..* " exit } From c8a32b578c8785601d417b85548c5e6a77ce012b Mon Sep 17 00:00:00 2001 From: yfpeng Date: Tue, 14 Aug 2018 15:59:08 +0800 Subject: [PATCH 5/9] debug memory leak --- .../otter/canal/store/memory/MemoryEventStoreWithBuffer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java index 00ba2e71..b276832a 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java @@ -417,7 +417,7 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple if (batchMode.isMemSize()) { ackMemSize.addAndGet(memsize); // 尝试清空buffer中的内存，将ack之前的内存全部释放掉 - for (long index = sequence + 1; index < next; index++) { + for (long index = sequence + 1; index <= next; index++) { entries[getIndex(index)] = null;// 设置为null } } From 679fd464f4b18a25b3a152785da480ec03f93d56 Mon Sep 17 00:00:00 2001 From: winger Date: Tue, 14 Aug 2018 16:09:44 +0800 Subject: [PATCH 6/9] =?UTF-8?q?=E5=8D=87=E7=BA=A7Kafka=20=E7=89=88?= =?UTF-8?q?=E6=9C=AC=E8=87=B3=202.11=5F1.1.1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka-client/pom.xml | 2 +- kafka/pom.xml | 9 +++++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/kafka-client/pom.xml b/kafka-client/pom.xml index d814be52..b589bd16 100644 --- a/kafka-client/pom.xml +++ b/kafka-client/pom.xml @@ -24,7 +24,7 @@ org.apache.kafka kafka-clients - 0.9.0.1 + 1.1.1 diff --git a/kafka/pom.xml b/kafka/pom.xml index 8d3b35d7..f1f099ed 100644 --- a/kafka/pom.xml +++ b/kafka/pom.xml @@ -25,11 +25,10 @@ snakeyaml 1.17 - org.apache.kafka kafka_2.11 - 0.9.0.1 + 1.1.1 org.slf4j @@ -37,6 +36,12 @@ + + + org.jboss.netty + netty + 3.2.2.Final + From 47765526050851ff34f15eccf7e25d0e04b5c6c3 Mon Sep 17 00:00:00 2001 From: mcy Date: Wed, 15 Aug 2018 14:26:55 +0800 Subject: [PATCH 7/9] =?UTF-8?q?=E5=8A=A0=E4=B8=8A=20if=20(logger.isDebugEn?= =?UTF-8?q?abled())=20=E9=98=B2=E6=AD=A2message.toString=E6=AF=8F=E6=AC=A1?= =?UTF-8?q?=E9=83=BD=E4=BC=9A=E8=A2=AB=E8=BF=90=E8=A1=8C=EF=BC=8Cmessage?= =?UTF-8?q?=E8=BF=87=E5=A4=A7=E4=BC=9A=E5=8D=A1cpu?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/producer/CanalKafkaProducer.java | 6 ++++-- kafka/src/main/resources/kafka.yml | 4 ++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java index 4543aac1..0bb15308 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -21,7 +21,7 @@ import com.alibaba.otter.canal.protocol.Message; */ public class CanalKafkaProducer { - private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); + private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); private Producer producer; @@ -83,6 +83,8 @@ public class CanalKafkaProducer { record = new ProducerRecord(topic.getTopic(), message); } producer.send(record); - logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); + if (logger.isDebugEnabled()) { + logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); + } } } diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index d6ccc51a..2729fc27 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -3,7 +3,7 @@ retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 -# canal的批次大小，单位 k +# canal的批次大小，单位 k，量大建议改为1M canalBatchSize: 50 filterTransactionEntry: true @@ -13,7 +13,7 @@ canalDestinations: partition: # 一个destination可以对应多个topic # topics: -# - topics: example +# - topic: example # partition: From a1588e5f983dd06bfff24ab7c9e99d412f502935 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 15 Aug 2018 19:45:11 +0800 Subject: [PATCH 8/9] fixed merge code --- .../otter/canal/deployer/CanalController.java | 8 ++++++++ deployer/src/main/resources/canal.properties | 1 + .../canal/prometheus/PrometheusService.java | 12 ++++++++--- .../prometheus/impl/ParserCollector.java | 20 +++++++++---------- .../canal/prometheus/impl/StoreCollector.java | 6 ++++-- .../embedded/CanalServerWithEmbedded.java | 6 ++++++ .../otter/canal/spi/CanalMetricsService.java | 5 +++++ .../canal/spi/NopCanalMetricsService.java | 5 +++++ 8 files changed, 48 insertions(+), 15 deletions(-) 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 6b0fc934..aacace84 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 @@ -98,6 +98,14 @@ public class CanalController { port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT)); embededCanalServer = CanalServerWithEmbedded.instance(); embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator + try { + int metricsPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_METRICS_PULL_PORT)); + embededCanalServer.setMetricsPort(metricsPort); + } catch (NumberFormatException e) { + logger.info("No valid metrics server port found, use default 11112."); + embededCanalServer.setMetricsPort(11112); + } + String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY); if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) { canalServer = CanalServerWithNetty.instance(); diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 9083477d..1cefb7d9 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -4,6 +4,7 @@ canal.id= 1 canal.ip= canal.port=11111 +canal.metrics.pull.port=11112 canal.zkServers= # flush data to zk canal.zookeeper.flush.period = 1000 diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java index 485f6728..18bf63f7 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java @@ -22,6 +22,7 @@ public class PrometheusService implements CanalMetricsService { private static final Logger logger = LoggerFactory.getLogger(PrometheusService.class); private final CanalInstanceExports instanceExports; private volatile boolean running = false; + private int port; private HTTPServer server; private final ClientInstanceProfiler clientProfiler; @@ -41,9 +42,9 @@ public class PrometheusService implements CanalMetricsService { @Override public void initialize() { try { - //TODO 1.Configurable port - //TODO 2.Https - server = new HTTPServer(11112); + logger.info("Start prometheus HTTPServer on port {}.", port); + //TODO 2.Https? + server = new HTTPServer(port); } catch (IOException e) { logger.warn("Unable to start prometheus HTTPServer.", e); return; @@ -112,4 +113,9 @@ public class PrometheusService implements CanalMetricsService { logger.info("Unregister metrics for destination {}.", instance.getDestination()); } + @Override + public void setServerPort(int port) { + this.port = port; + } + } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java index 73102795..867e1e89 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java @@ -1,15 +1,10 @@ package com.alibaba.otter.canal.prometheus.impl; -import com.alibaba.otter.canal.instance.core.CanalInstance; -import com.alibaba.otter.canal.parse.CanalEventParser; -import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; -import com.alibaba.otter.canal.prometheus.InstanceRegistry; -import com.google.common.base.Preconditions; +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST; +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; import io.prometheus.client.Collector; import io.prometheus.client.CounterMetricFamily; import io.prometheus.client.GaugeMetricFamily; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Arrays; @@ -19,9 +14,14 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; -import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST; -import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; -import static java.util.concurrent.TimeUnit.NANOSECONDS; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.instance.core.CanalInstance; +import com.alibaba.otter.canal.parse.CanalEventParser; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; +import com.alibaba.otter.canal.prometheus.InstanceRegistry; +import com.google.common.base.Preconditions; /** * @author Chuanyi Li diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java index 3be87c3e..d345602c 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java @@ -40,7 +40,7 @@ public class StoreCollector extends Collector implements InstanceRegistry { private static final String PRODUCE_MEM_HELP = "Produced mem bytes of canal instance"; private static final String CONSUME_MEM_HELP = "Consumed mem bytes of canal instance"; private final ConcurrentMap instances = new ConcurrentHashMap(); - private final List storeLabelsList = Arrays.asList(DEST, "batchMode"); + private final List storeLabelsList = Arrays.asList(DEST, "batchMode", "size"); private StoreCollector() {} @@ -100,7 +100,8 @@ public class StoreCollector extends Collector implements InstanceRegistry { holder.putSeq = memStore.getPutSequence(); holder.ackSeq = memStore.getAckSequence(); holder.destLabelValues = Collections.singletonList(destination); - holder.storeLabelValues = Arrays.asList(destination, memStore.getBatchMode().name()); + holder.size = memStore.getBufferSize(); + holder.storeLabelValues = Arrays.asList(destination, holder.batchMode.name(), Integer.toString(holder.size)); Preconditions.checkNotNull(holder.batchMode); Preconditions.checkNotNull(holder.putSeq); Preconditions.checkNotNull(holder.ackSeq); @@ -128,6 +129,7 @@ public class StoreCollector extends Collector implements InstanceRegistry { private BatchMode batchMode; private AtomicLong putMemSize; private AtomicLong ackMemSize; + private int size; private List destLabelValues; private List storeLabelValues; } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index ccf21020..9ac5a834 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -44,6 +44,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C private Map canalInstances; // private Map lastRollbackPostions; private CanalInstanceGenerator canalInstanceGenerator; + private int metricsPort; private CanalMetricsService metrics = NopCanalMetricsService.NOP; private static class SingletonHolder { @@ -64,6 +65,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C super.start(); // 如果存在provider,则启动metrics service loadCanalMetrics(); + metrics.setServerPort(metricsPort); metrics.initialize(); canalInstances = MigrateMap.makeComputingMap(new Function() { @@ -529,4 +531,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C this.canalInstanceGenerator = canalInstanceGenerator; } + public void setMetricsPort(int metricsPort) { + this.metricsPort = metricsPort; + } + } diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java index 6d823be8..a4528800 100644 --- a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java +++ b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java @@ -39,4 +39,9 @@ public interface CanalMetricsService { */ void unregister(CanalInstance instance); + /** + * @param port server port for pull + */ + void setServerPort(int port); + } diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java b/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java index 0d6ad088..45926348 100644 --- a/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java +++ b/server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java @@ -35,4 +35,9 @@ public class NopCanalMetricsService implements CanalMetricsService { public void unregister(CanalInstance instance) { } + + @Override + public void setServerPort(int port) { + + } } From e879216fa4448c82931145dcb96a22280196fa01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 17 Aug 2018 10:40:51 +0800 Subject: [PATCH 9/9] fixed compiler --- .../java/com/alibaba/otter/canal/deployer/CanalConstants.java | 1 + .../otter/canal/store/memory/MemoryEventStoreWithBuffer.java | 3 +++ 2 files changed, 4 insertions(+) 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 b328feb7..c7a90d30 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 @@ -15,6 +15,7 @@ public class CanalConstants { public static final String CANAL_ID = ROOT + "." + "id"; public static final String CANAL_IP = ROOT + "." + "ip"; public static final String CANAL_PORT = ROOT + "." + "port"; + public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port"; public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers"; public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty"; diff --git a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java index b276832a..a1ff4033 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java @@ -545,6 +545,9 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple } // ================ setter / getter ================== + public int getBufferSize() { + return this.bufferSize; + } public void setBufferSize(int bufferSize) { this.bufferSize = bufferSize;