diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 2a24b66f..b5284dec 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -69,6 +69,10 @@ canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destinat canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; canal.instance.tsdb.dbUsername=canal canal.instance.tsdb.dbPassword=canal +# dump snapshot interval, default 24 hour +canal.instance.tsdb.snapshot.interval=24 +# purge snapshot expire , default 360 hour(15 days) +canal.instance.tsdb.snapshot.expire=360 # rds oss binlog account canal.instance.rds.accesskey = diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index fe9cc7a5..f8c19bd0 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -176,6 +176,8 @@ + + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 23d8e4bd..5524f73f 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -161,6 +161,8 @@ + + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index f785a46e..8fdfb178 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -149,6 +149,8 @@ + + diff --git a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml index 7445bf69..58957c28 100644 --- a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml +++ b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml @@ -36,10 +36,10 @@ - + \ No newline at end of file diff --git a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml index 79aa25fe..2ee0f8f4 100644 --- a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml +++ b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml @@ -42,10 +42,10 @@ where destination=#destination# - + 0 ]]> \ No newline at end of file diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index dff18278..154cd53f 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -296,6 +296,12 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { mysqlEventParser.setFilterTableError(parameters.getFilterTableError()); mysqlEventParser.setIsGTIDMode(BooleanUtils.toBoolean(parameters.getGtidEnable())); // tsdb + if (parameters.getTsdbSnapshotInterval() != null) { + mysqlEventParser.setTsdbSnapshotInterval(parameters.getTsdbSnapshotInterval()); + } + if (parameters.getTsdbSnapshotExpire() != null) { + mysqlEventParser.setTsdbSnapshotExpire(parameters.getTsdbSnapshotExpire()); + } boolean tsdbEnable = BooleanUtils.toBoolean(parameters.getTsdbEnable()); if (tsdbEnable) { mysqlEventParser.setEnableTsdb(tsdbEnable); diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java index ca5e1af7..af3841dd 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java @@ -98,6 +98,8 @@ public class CanalParameter implements Serializable { private String tsdbJdbcUrl; private String tsdbJdbcUserName; private String tsdbJdbcPassword; + private Integer tsdbSnapshotInterval = 24; + private Integer tsdbSnapshotExpire = 360; private String rdsAccesskey; private String rdsSecretkey; private String rdsInstanceId; @@ -964,6 +966,22 @@ public class CanalParameter implements Serializable { this.memoryStorageRawEntry = memoryStorageRawEntry; } + public Integer getTsdbSnapshotInterval() { + return tsdbSnapshotInterval; + } + + public void setTsdbSnapshotInterval(Integer tsdbSnapshotInterval) { + this.tsdbSnapshotInterval = tsdbSnapshotInterval; + } + + public Integer getTsdbSnapshotExpire() { + return tsdbSnapshotExpire; + } + + public void setTsdbSnapshotExpire(Integer tsdbSnapshotExpire) { + this.tsdbSnapshotExpire = tsdbSnapshotExpire; + } + public String toString() { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } 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 00e302e6..8050186d 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 @@ -26,6 +26,8 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected TableMetaTSDBFactory tableMetaTSDBFactory = new DefaultTableMetaTSDBFactory(); protected boolean enableTsdb = false; + protected int tsdbSnapshotInterval = 24; + protected int tsdbSnapshotExpire = 360; protected String tsdbSpringXml; protected TableMetaTSDB tableMetaTSDB; @@ -210,4 +212,20 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { return this.receivedBinlogBytes; } + public int getTsdbSnapshotInterval() { + return tsdbSnapshotInterval; + } + + public void setTsdbSnapshotInterval(int tsdbSnapshotInterval) { + this.tsdbSnapshotInterval = tsdbSnapshotInterval; + } + + public int getTsdbSnapshotExpire() { + return tsdbSnapshotExpire; + } + + public void setTsdbSnapshotExpire(int tsdbSnapshotExpire) { + this.tsdbSnapshotExpire = tsdbSnapshotExpire; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java index ea6b0365..4f6ace4f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java @@ -55,6 +55,8 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements ((DatabaseTableMeta) tableMetaTSDB).setConnection(metaConnection); ((DatabaseTableMeta) tableMetaTSDB).setFilter(eventFilter); ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter); + ((DatabaseTableMeta) tableMetaTSDB).setSnapshotInterval(tsdbSnapshotInterval); + ((DatabaseTableMeta) tableMetaTSDB).setSnapshotExpire(tsdbSnapshotExpire); } tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); 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 cf102d3d..7f83b6d2 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 @@ -121,6 +121,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE ((DatabaseTableMeta) tableMetaTSDB).setConnection(metaConnection); ((DatabaseTableMeta) tableMetaTSDB).setFilter(eventFilter); ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter); + ((DatabaseTableMeta) tableMetaTSDB).setSnapshotInterval(tsdbSnapshotInterval); + ((DatabaseTableMeta) tableMetaTSDB).setSnapshotExpire(tsdbSnapshotExpire); } tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index bb2af45f..d7b2f7d2 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -44,19 +44,21 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition; */ public class DatabaseTableMeta implements TableMetaTSDB { - public static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L); - private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class); - private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); - private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation"); + public static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L); + private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class); + private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); + private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation"); private String destination; private MemoryTableMeta memoryTableMeta; - private MysqlConnection connection; // 查询meta信息的链接 + private MysqlConnection connection; // 查询meta信息的链接 private CanalEventFilter filter; private CanalEventFilter blackFilter; private EntryPosition lastPosition; private ScheduledExecutorService scheduler; private MetaHistoryDAO metaHistoryDAO; private MetaSnapshotDAO metaSnapshotDAO; + private int snapshotInterval = 24; + private int snapshotExpire = 360; public DatabaseTableMeta(){ @@ -77,18 +79,27 @@ public class DatabaseTableMeta implements TableMetaTSDB { }); // 24小时生成一份snapshot - scheduler.scheduleWithFixedDelay(new Runnable() { + if (snapshotInterval > 0) { + scheduler.scheduleWithFixedDelay(new Runnable() { - @Override - public void run() { - try { - MDC.put("destination", destination); - applySnapshotToDB(lastPosition, false); - } catch (Throwable e) { - logger.error("scheudle applySnapshotToDB faield", e); + @Override + public void run() { + try { + MDC.put("destination", destination); + applySnapshotToDB(lastPosition, false); + } catch (Throwable e) { + logger.error("scheudle applySnapshotToDB faield", e); + } + + try { + MDC.put("destination", destination); + snapshotExpire((int) TimeUnit.HOURS.toSeconds(snapshotExpire)); + } catch (Throwable e) { + logger.error("scheudle snapshotExpire faield", e); + } } - } - }, 24, 24, TimeUnit.HOURS); + }, snapshotInterval, snapshotInterval, TimeUnit.HOURS); + } return true; } @@ -461,6 +472,10 @@ public class DatabaseTableMeta implements TableMetaTSDB { return true; } + private int snapshotExpire(int expireTimestamp) { + return metaSnapshotDAO.deleteByTimestamp(destination, expireTimestamp); + } + public void setConnection(MysqlConnection connection) { this.connection = connection; } @@ -489,6 +504,22 @@ public class DatabaseTableMeta implements TableMetaTSDB { this.blackFilter = blackFilter; } + public int getSnapshotInterval() { + return snapshotInterval; + } + + public void setSnapshotInterval(int snapshotInterval) { + this.snapshotInterval = snapshotInterval; + } + + public int getSnapshotExpire() { + return snapshotExpire; + } + + public void setSnapshotExpire(int snapshotExpire) { + this.snapshotExpire = snapshotExpire; + } + public MysqlConnection getConnection() { return connection; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java index 6ec51614..c214d88f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; -import java.text.SimpleDateFormat; -import java.util.Date; import java.util.HashMap; import java.util.List; @@ -37,13 +35,12 @@ public class MetaHistoryDAO extends MetaBaseDAO { /** * 删除interval秒之前的数据 */ - public Integer deleteByGmtModified(int interval) { + public Integer deleteByTimestamp(String destination, int interval) { HashMap params = Maps.newHashMapWithExpectedSize(2); long timestamp = System.currentTimeMillis() - interval * 1000; - Date date = new Date(timestamp); - SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); - params.put("timestamp", format.format(date)); - return getSqlMapClientTemplate().delete("meta_history.deleteByGmtModified", params); + params.put("timestamp", timestamp); + params.put("destination", destination); + return getSqlMapClientTemplate().delete("meta_history.deleteByTimestamp", params); } protected void initDao() throws Exception { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java index f525031a..ff79b107 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; -import java.text.SimpleDateFormat; -import java.util.Date; import java.util.HashMap; import com.google.common.collect.Maps; @@ -40,13 +38,12 @@ public class MetaSnapshotDAO extends MetaBaseDAO { /** * 删除interval秒之前的数据 */ - public Integer deleteByGmtModified(int interval) { + public Integer deleteByTimestamp(String destination, int interval) { HashMap params = Maps.newHashMapWithExpectedSize(2); long timestamp = System.currentTimeMillis() - interval * 1000; - Date date = new Date(timestamp); - SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); - params.put("timestamp", format.format(date)); - return getSqlMapClientTemplate().delete("meta_snapshot.deleteByGmtModified", params); + params.put("timestamp", timestamp); + params.put("destination", destination); + return getSqlMapClientTemplate().delete("meta_snapshot.deleteByTimestamp", params); } protected void initDao() throws Exception { diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java index b199a326..cea6c087 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java @@ -16,7 +16,7 @@ import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO; * Created by wanshao Date: 2017/9/20 Time: 下午5:00 **/ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" }) +@ContextConfiguration(locations = { "/tsdb/h2-tsdb.xml" }) public class MetaHistoryDAOTest { @Resource @@ -24,7 +24,22 @@ public class MetaHistoryDAOTest { @Test public void testSimple() { - List metaHistoryDOList = metaHistoryDAO.findByTimestamp("test", 0L, 0L); + MetaHistoryDO historyDO = new MetaHistoryDO(); + historyDO.setDestination("test"); + historyDO.setBinlogFile("000001"); + historyDO.setBinlogOffest(4L); + historyDO.setBinlogMasterId("1"); + historyDO.setBinlogTimestamp(System.currentTimeMillis() - 7300 * 1000); + historyDO.setSqlSchema("test"); + historyDO.setUseSchema("test"); + historyDO.setSqlTable("testTable"); + historyDO.setSqlTable("drop table testTable"); + metaHistoryDAO.insert(historyDO); + + int count = metaHistoryDAO.deleteByTimestamp("test", 7200); + System.out.println(count); + + List metaHistoryDOList = metaHistoryDAO.findByTimestamp("test", 0L, System.currentTimeMillis()); for (MetaHistoryDO metaHistoryDO : metaHistoryDOList) { System.out.println(metaHistoryDO.getId()); } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaSnapshotDAOTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaSnapshotDAOTest.java new file mode 100644 index 00000000..f503e4b2 --- /dev/null +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaSnapshotDAOTest.java @@ -0,0 +1,41 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; + +import javax.annotation.Resource; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDO; + +/** + * Created by wanshao Date: 2017/9/20 Time: 下午5:00 + **/ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(locations = { "/tsdb/h2-tsdb.xml" }) +public class MetaSnapshotDAOTest { + + @Resource + MetaSnapshotDAO metaSnapshotDAO; + + @Test + public void testSimple() { + MetaSnapshotDO metaSnapshotDO = new MetaSnapshotDO(); + metaSnapshotDO.setDestination("test"); + metaSnapshotDO.setBinlogFile("000001"); + metaSnapshotDO.setBinlogOffest(4L); + metaSnapshotDO.setBinlogMasterId("1"); + metaSnapshotDO.setBinlogTimestamp(System.currentTimeMillis() - 7300 * 1000); + metaSnapshotDO.setData("test"); + metaSnapshotDAO.insert(metaSnapshotDO); + + MetaSnapshotDO snapshotDO = metaSnapshotDAO.findByTimestamp("test", System.currentTimeMillis()); + System.out.println(snapshotDO); + + int count = metaSnapshotDAO.deleteByTimestamp("test", 7200); + System.out.println(count); + } + +} diff --git a/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml index 7445bf69..58957c28 100644 --- a/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml +++ b/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml @@ -36,10 +36,10 @@ - + \ No newline at end of file diff --git a/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml index 7ba0a435..60827c6c 100644 --- a/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml +++ b/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml @@ -54,10 +54,10 @@ where destination=#destination# - + 0 ]]> \ No newline at end of file