Merge pull request #827 from lcybo/metrics_support

Merge master updates

deployer/pom.xml

@@ -18,12 +18,12 @@
 		</dependency>
 
 		<!-- 这里指定runtime的metrics provider-->
-		<!--<dependency>-->
-			<!--<groupId>com.alibaba.otter</groupId>-->
-			<!--<artifactId>canal.prometheus</artifactId>-->
-			<!--<version>${project.version}</version>-->
-			<!--<scope>runtime</scope>-->
-		<!--</dependency>-->
+		<dependency>
+			<groupId>com.alibaba.otter</groupId>
+			<artifactId>canal.prometheus</artifactId>
+			<version>${project.version}</version>
+			<scope>runtime</scope>
+		</dependency>
 	</dependencies>
 	
 	<build>

deployer/src/main/bin/metrics_env.sh

@@ -1,15 +0,0 @@
-#!/bin/bash
-# Additional line arg for current prometheus solution
-case "`uname`" in
-Linux)
-    bin_abs_path=$(readlink -f $(dirname $0))
-	;;
-*)
-	bin_abs_path=`cd $(dirname $0); pwd`
-	;;
-esac
-base=${bin_abs_path}/..
-if [ $(ls $base/lib/aspectjweaver*.jar | wc -l) -eq 1 ]; then
-    WEAVER=$(ls $base/lib/aspectjweaver*.jar)
-    METRICS_OPTS=" -javaagent:"${WEAVER}" "
-fi

deployer/src/main/bin/startup.sh

@@ -94,12 +94,7 @@ then
 	echo LOG CONFIGURATION : $logback_configurationFile
 	echo canal conf : $canal_conf 
 	echo CLASSPATH :$CLASSPATH
-#   metrics support options
-#	if [ -x $base/bin/metrics_env.sh ]; then
-#	    . $base/bin/metrics_env.sh
-#	    echo METRICS_OPTS $METRICS_OPTS
-#	fi
-	$JAVA $JAVA_OPTS $METRICS_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.deployer.CanalLauncher 1>>$base/logs/canal/canal.log 2>&1 &
+	$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $CANAL_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.deployer.CanalLauncher 1>>$base/logs/canal/canal.log 2>&1 &
 	echo $! > $base/bin/canal.pid 
 	
 	echo "cd to $current_path for continue"

example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java

@@ -1,9 +1,9 @@
 package com.alibaba.otter.canal.example;
 
+import java.io.UnsupportedEncodingException;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.SystemUtils;
@@ -254,7 +254,18 @@ public class AbstractCanalClientTest {
     protected void printColumn(List<Column> columns) {
         for (Column column : columns) {
             StringBuilder builder = new StringBuilder();
-            builder.append(column.getName() + " : " + column.getValue());
+            try {
+                if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
+                    || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY")) {
+                    // get value bytes
+                    builder.append(column.getName() + " : "
+                                   + new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8"));
+                } else {
+                    builder.append(column.getName() + " : " + column.getValue());
+                }
+            } catch (UnsupportedEncodingException e) {
+            }
+
             builder.append("    type=" + column.getMysqlType());
             if (column.getUpdated()) {
                 builder.append("    update=" + column.getUpdated());

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java

@@ -95,6 +95,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
     protected int                                    parallelBufferSize         = 256;                                     // 必须为2的幂
     protected MultiStageCoprocessor                  multiStageCoprocessor;
 
+
     protected abstract BinlogParser buildParser();
 
     protected abstract ErosaConnection buildErosaConnection();
@@ -615,4 +616,5 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         this.parallelBufferSize = parallelBufferSize;
     }
 
+
 }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBuffer.java

@@ -80,6 +80,11 @@ public class EventTransactionBuffer extends AbstractCanalLifeCycle {
                     flush();
                 }
                 break;
+            case HEARTBEAT:
+                // master过来的heartbeat，说明binlog已经读完了，是idle状态
+                put(entry);
+                flush();
+                break;
             default:
                 break;
         }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java

@@ -1,6 +1,7 @@
 package com.alibaba.otter.canal.parse.inbound.mysql;
 
 import java.nio.charset.Charset;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -21,23 +22,24 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition;
 
 public abstract class AbstractMysqlEventParser extends AbstractEventParser {
 
-    protected final Logger         logger                  = LoggerFactory.getLogger(this.getClass());
-    protected static final long    BINLOG_START_OFFEST     = 4L;
+    protected final Logger         logger                    = LoggerFactory.getLogger(this.getClass());
+    protected static final long    BINLOG_START_OFFEST       = 4L;
 
-    protected TableMetaTSDBFactory tableMetaTSDBFactory    = new DefaultTableMetaTSDBFactory();
-    protected boolean              enableTsdb              = false;
+    protected TableMetaTSDBFactory tableMetaTSDBFactory      = new DefaultTableMetaTSDBFactory();
+    protected boolean              enableTsdb                = false;
     protected String               tsdbSpringXml;
     protected TableMetaTSDB        tableMetaTSDB;
 
     // 编码信息
-    protected byte                 connectionCharsetNumber = (byte) 33;
-    protected Charset              connectionCharset       = Charset.forName("UTF-8");
-    protected boolean              filterQueryDcl          = false;
-    protected boolean              filterQueryDml          = false;
-    protected boolean              filterQueryDdl          = false;
-    protected boolean              filterRows              = false;
-    protected boolean              filterTableError        = false;
-    protected boolean              useDruidDdlFilter       = true;
+    protected byte                 connectionCharsetNumber   = (byte) 33;
+    protected Charset              connectionCharset         = Charset.forName("UTF-8");
+    protected boolean              filterQueryDcl            = false;
+    protected boolean              filterQueryDml            = false;
+    protected boolean              filterQueryDdl            = false;
+    protected boolean              filterRows                = false;
+    protected boolean              filterTableError          = false;
+    protected boolean              useDruidDdlFilter         = true;
+    private final AtomicLong       eventsPublishBlockingTime = new AtomicLong(0L);
 
     protected BinlogParser buildParser() {
         LogEventConvert convert = new LogEventConvert();
@@ -131,11 +133,13 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
     }
 
     protected MultiStageCoprocessor buildMultiStageCoprocessor() {
-        return new MysqlMultiStageCoprocessor(parallelBufferSize,
-            parallelThreadSize,
-            (LogEventConvert) binlogParser,
-            transactionBuffer,
-            destination);
+        MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize,
+                parallelThreadSize,
+                (LogEventConvert) binlogParser,
+                transactionBuffer,
+                destination);
+        mysqlMultiStageCoprocessor.setEventsPublishBlockingTime(eventsPublishBlockingTime);
+        return mysqlMultiStageCoprocessor;
     }
 
     // ============================ setter / getter =========================
@@ -200,6 +204,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
         }
     }
 
+    public AtomicLong getEventsPublishBlockingTime() {
+        return this.eventsPublishBlockingTime;
+    }
+    
     public void setTableMetaTSDBFactory(TableMetaTSDBFactory tableMetaTSDBFactory) {
         this.tableMetaTSDBFactory = tableMetaTSDBFactory;
     }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java

@@ -7,6 +7,7 @@ import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -37,18 +38,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;
 
     public MysqlConnection(){
     }
@@ -124,6 +127,7 @@ public class MysqlConnection implements ErosaConnection {
         decoder.handle(LogEvent.XID_EVENT);
         LogContext context = new LogContext();
         while (fetcher.fetch()) {
+            accumulateReceivedBytes(fetcher.limit());
             LogEvent event = null;
             event = decoder.decode(fetcher, context);
 
@@ -146,6 +150,7 @@ public class MysqlConnection implements ErosaConnection {
         LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
         LogContext context = new LogContext();
         while (fetcher.fetch()) {
+            accumulateReceivedBytes(fetcher.limit());
             LogEvent event = null;
             event = decoder.decode(fetcher, context);
 
@@ -174,6 +179,7 @@ public class MysqlConnection implements ErosaConnection {
             LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
             LogContext context = new LogContext();
             while (fetcher.fetch()) {
+                accumulateReceivedBytes(fetcher.limit());
                 LogEvent event = null;
                 event = decoder.decode(fetcher, context);
 
@@ -204,6 +210,7 @@ public class MysqlConnection implements ErosaConnection {
         try {
             fetcher.start(connector.getChannel());
             while (fetcher.fetch()) {
+                accumulateReceivedBytes(fetcher.limit());
                 LogBuffer buffer = fetcher.duplicate();
                 fetcher.consume(fetcher.limit());
                 if (!coprocessor.publish(buffer)) {
@@ -230,6 +237,7 @@ public class MysqlConnection implements ErosaConnection {
         try {
             fetcher.start(connector.getChannel());
             while (fetcher.fetch()) {
+                accumulateReceivedBytes(fetcher.limit());
                 LogBuffer buffer = fetcher.duplicate();
                 fetcher.consume(fetcher.limit());
                 if (!coprocessor.publish(buffer)) {
@@ -334,7 +342,6 @@ public class MysqlConnection implements ErosaConnection {
      * <li>net_read_timeout</li>
      * </ol>
      * 
-     * @param channel
      * @throws IOException
      */
     private void updateSettings() throws IOException {
@@ -453,6 +460,14 @@ public class MysqlConnection implements ErosaConnection {
         }
     }
 
+    private void accumulateReceivedBytes(long x) {
+        if (receivedBinlogBytes != null) {
+            receivedBinlogBytes.addAndGet(x);
+        }
+    }
+
+
+
     public static enum BinlogFormat {
 
         STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED");
@@ -592,4 +607,8 @@ public class MysqlConnection implements ErosaConnection {
         this.authInfo = authInfo;
     }
 
+    public void setReceivedBinlogBytes(AtomicLong receivedBinlogBytes) {
+        this.receivedBinlogBytes = receivedBinlogBytes;
+    }
+
 }

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

@@ -68,8 +68,11 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     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);
 
     protected ErosaConnection buildErosaConnection() {
         return buildMysqlConnection(this.runningInfo);
@@ -313,6 +316,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         connection.getConnector().setSendBufferSize(sendBufferSize);
         connection.getConnector().setSoTimeout(defaultConnectionTimeoutInSeconds * 1000);
         connection.setCharset(connectionCharset);
+        connection.setReceivedBinlogBytes(receivedBinlogBytes);
         // 随机生成slaveId
         if (this.slaveId <= 0) {
             this.slaveId = generateUniqueServerId();
@@ -511,7 +515,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     private Long findTransactionBeginPosition(ErosaConnection mysqlConnection, final EntryPosition entryPosition)
                                                                                                                  throws IOException {
         // 针对开始的第一条为非Begin记录，需要从该binlog扫描
-        final AtomicLong preTransactionStartPosition = new AtomicLong(0L);
+        final java.util.concurrent.atomic.AtomicLong preTransactionStartPosition = new java.util.concurrent.atomic.AtomicLong(0L);
         mysqlConnection.reconnect();
         mysqlConnection.seek(entryPosition.getJournalName(), 4L, new SinkFunction<LogEvent>() {
 
@@ -910,4 +914,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         this.dumpErrorCountThreshold = dumpErrorCountThreshold;
     }
 
+
+
+    public AtomicLong getReceivedBinlogBytes() {
+        return this.receivedBinlogBytes;
+    }
+
 }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.LockSupport;
 
 import org.apache.commons.lang.StringUtils;
@@ -52,6 +53,7 @@ import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
  */
 public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implements MultiStageCoprocessor {
 
+    private static final int             maxFullTimes               = 10;
     private LogEventConvert              logEventConvert;
     private EventTransactionBuffer       transactionBuffer;
     private ErosaConnection              connection;
@@ -63,6 +65,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
     private ExecutorService              stageExecutor;
     private String                       destination;
     private volatile CanalParseException exception;
+    private AtomicLong                   eventsPublishBlockingTime;
 
     public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
                                       EventTransactionBuffer transactionBuffer, String destination){
@@ -161,6 +164,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
             throw exception;
         }
         boolean interupted = false;
+        long blockingStart = 0L;
+        int fullTimes = 0;
         do {
             try {
                 long next = disruptorMsgBuffer.tryNext();
@@ -170,16 +175,39 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
                     event.setBinlogFileName(binlogFileName);
                 }
                 disruptorMsgBuffer.publish(next);
+                if (fullTimes > 0) {
+                    eventsPublishBlockingTime.addAndGet(System.nanoTime() - blockingStart);
+                }
                 break;
             } catch (InsufficientCapacityException e) {
+                if (fullTimes == 0) {
+                    blockingStart = System.nanoTime();
+                }
                 // park
-                LockSupport.parkNanos(1L);
+                //LockSupport.parkNanos(1L);
+                applyWait(++fullTimes);
                 interupted = Thread.interrupted();
+                if (fullTimes % 1000 == 0) {
+                    long nextStart = System.nanoTime();
+                    eventsPublishBlockingTime.addAndGet(nextStart - blockingStart);
+                    blockingStart = nextStart;
+                }
             }
         } while (!interupted && isStart());
         return isStart();
     }
 
+    // 处理无数据的情况，避免空循环挂死
+    private void applyWait(int fullTimes) {
+        int newFullTimes = fullTimes > maxFullTimes ? maxFullTimes : fullTimes;
+        if (fullTimes <= 3) { // 3次以内
+            Thread.yield();
+        } else { // 超过3次，最多只sleep 1ms
+            LockSupport.parkNanos(100 * 1000L * newFullTimes);
+        }
+
+    }
+
     @Override
     public void reset() {
         if (isStart()) {
@@ -189,6 +217,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
         start();
     }
 
+
     private class SimpleParserStage implements EventHandler<MessageEvent>, LifecycleAware {
 
         private LogDecoder decoder;
@@ -427,4 +456,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
         this.connection = connection;
     }
 
+    public void setEventsPublishBlockingTime(AtomicLong eventsPublishBlockingTime) {
+        this.eventsPublishBlockingTime = eventsPublishBlockingTime;
+    }
+
 }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java

@@ -10,6 +10,7 @@ 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;
@@ -40,22 +41,7 @@ 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.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;
 
 /**
@@ -144,6 +130,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 return parseRandLogEvent((RandLogEvent) logEvent);
             case LogEvent.GTID_LOG_EVENT:
                 return parseGTIDLogEvent((GtidLogEvent) logEvent);
+            case LogEvent.HEARTBEAT_LOG_EVENT:
+                return parseHeartbeatLogEvent((HeartbeatLogEvent) logEvent);
             default:
                 break;
         }
@@ -158,6 +146,15 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         }
     }
 
+    private Entry parseHeartbeatLogEvent(HeartbeatLogEvent logEvent) {
+        Header.Builder headerBuilder = Header.newBuilder();
+        headerBuilder.setEventType(EventType.MHEARTBEAT);
+        Entry.Builder entryBuilder = Entry.newBuilder();
+        entryBuilder.setHeader(headerBuilder.build());
+        entryBuilder.setEntryType(EntryType.HEARTBEAT);
+        return entryBuilder.build();
+    }
+
     private Entry parseGTIDLogEvent(GtidLogEvent logEvent) {
         LogHeader logHeader = logEvent.getHeader();
         String value = logEvent.getSid().toString() + ":" + logEvent.getGno();
@@ -759,6 +756,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                         } else {
                             // byte数组，直接使用iso-8859-1保留对应编码，浪费内存
                             columnBuilder.setValue(new String((byte[]) value, ISO_8859_1));
+                            // columnBuilder.setValueBytes(ByteString.copyFrom((byte[])
+                            // value));
                             javaType = Types.BLOB;
                         }
                         break;

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/NoStorageTest.java

@@ -0,0 +1,38 @@
+package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta;
+
+import com.alibaba.otter.canal.parse.inbound.TableMeta;
+import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection;
+import com.alibaba.otter.canal.protocol.position.EntryPosition;
+import org.junit.Test;
+
+import java.net.InetSocketAddress;
+import java.util.Date;
+
+public class NoStorageTest {
+    final String DBNAME = "testdb";
+    final String TBNAME = "testtb";
+    final String DDL = "CREATE TABLE `testtb` (\n" +
+            "   `id` int(11) NOT NULL AUTO_INCREMENT,\n" +
+            "   `name` varchar(2048) DEFAULT NULL,\n" +
+            "   `datachange_lasttime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最晚更新时间',\n" +
+            "   `otter_testcol` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol1` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol2` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol3` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol4` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol5` varchar(45) DEFAULT NULL,\n" +
+            "   PRIMARY KEY (`id`)\n" +
+            " ) ENGINE=InnoDB AUTO_INCREMENT=58333898 DEFAULT CHARSET=utf8mb4";
+    @Test
+    public void nostorage() {
+        MysqlConnection connection = new MysqlConnection(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
+        TableMetaCacheWithStorage tableMetaCacheWithStorage = new TableMetaCacheWithStorage(connection, null);
+        EntryPosition entryPosition = new EntryPosition();
+        entryPosition.setTimestamp(new Date().getTime());
+        String fullTableName = DBNAME + "." + TBNAME;
+        tableMetaCacheWithStorage.apply(entryPosition, fullTableName, DDL, null);
+        entryPosition.setTimestamp(new Date().getTime() + 1000L);
+        TableMeta result = tableMetaCacheWithStorage.getTableMeta(DBNAME, TBNAME, false, entryPosition);
+        assert result.getDdl().equalsIgnoreCase(DDL);
+    }
+}

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/StorageTest.java

@@ -0,0 +1,77 @@
+package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta;
+
+import com.alibaba.otter.canal.parse.inbound.TableMeta;
+import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection;
+import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql.MySqlTableMetaCallback;
+import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql.MySqlTableMetaStorageFactory;
+import com.alibaba.otter.canal.protocol.position.EntryPosition;
+import com.alibaba.otter.canal.protocol.position.Position;
+import org.junit.Test;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+public class StorageTest {
+
+    final String DBNAME = "testdb";
+    final String TBNAME = "testtb";
+    final String DDL = "CREATE TABLE `testtb` (\n" +
+            "   `id` int(11) NOT NULL AUTO_INCREMENT,\n" +
+            "   `name` varchar(2048) DEFAULT NULL,\n" +
+            "   `datachange_lasttime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最晚更新时间',\n" +
+            "   `otter_testcol` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol1` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol2` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol3` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol4` varchar(45) DEFAULT NULL,\n" +
+            "   `otter_testcol5` varchar(45) DEFAULT NULL,\n" +
+            "   PRIMARY KEY (`id`)\n" +
+            " ) ENGINE=InnoDB AUTO_INCREMENT=58333898 DEFAULT CHARSET=utf8mb4";
+
+    @Test
+    public void storage() {
+
+        MySqlTableMetaStorageFactory factory = new MySqlTableMetaStorageFactory(new MySqlTableMetaCallback() {
+            @Override
+            public void save(String dbAddress, String schema, String table, String ddl, Long timestamp) {
+
+            }
+
+            @Override
+            public List<TableMetaEntry> fetch(String dbAddress, String dbName) {
+                TableMetaEntry tableMeta = new TableMetaEntry();
+                tableMeta.setSchema(DBNAME);
+                tableMeta.setTable(TBNAME);
+                tableMeta.setDdl(DDL);
+                tableMeta.setTimestamp(new Date().getTime());
+                List<TableMetaEntry> entries = new ArrayList<TableMetaEntry>();
+                entries.add(tableMeta);
+                return entries;
+            }
+
+            @Override
+            public List<TableMetaEntry> fetch(String dbAddress, String dbName, String tableName) {
+                TableMetaEntry tableMeta = new TableMetaEntry();
+                tableMeta.setSchema(DBNAME);
+                tableMeta.setTable(TBNAME);
+                tableMeta.setDdl(DDL);
+                tableMeta.setTimestamp(new Date().getTime());
+                List<TableMetaEntry> entries = new ArrayList<TableMetaEntry>();
+                entries.add(tableMeta);
+                return entries;
+            }
+        }, DBNAME);
+        MysqlConnection connection = new MysqlConnection(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
+        TableMetaCacheWithStorage tableMetaCacheWithStorage = new TableMetaCacheWithStorage(connection, factory.getTableMetaStorage());
+        EntryPosition entryPosition = new EntryPosition();
+        entryPosition.setTimestamp(new Date().getTime());
+        String fullTableName = DBNAME + "." + TBNAME;
+        tableMetaCacheWithStorage.apply(entryPosition, fullTableName, DDL, null);
+
+        entryPosition.setTimestamp(new Date().getTime() + 1000L);
+        TableMeta result = tableMetaCacheWithStorage.getTableMeta(DBNAME, TBNAME, false, entryPosition);
+        assert result.getDdl().equalsIgnoreCase(DDL);
+    }
+}

prometheus/pom.xml

@@ -14,17 +14,6 @@
     <version>1.0.26-SNAPSHOT</version>
     <name>canal prometheus module for otter ${project.version}</name>
     <dependencies>
-        <!-- load time weaver-->
-        <dependency>
-            <groupId>org.aspectj</groupId>
-            <artifactId>aspectjrt</artifactId>
-            <version>1.8.9</version>
-        </dependency>
-        <dependency>
-            <groupId>org.aspectj</groupId>
-            <artifactId>aspectjweaver</artifactId>
-            <version>1.8.9</version>
-        </dependency>
         <dependency>
             <groupId>org.jctools</groupId>
             <artifactId>jctools-core</artifactId>

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalInstanceExports.java

@@ -1,12 +1,7 @@
 package com.alibaba.otter.canal.prometheus;
 
 import com.alibaba.otter.canal.instance.core.CanalInstance;
-import com.alibaba.otter.canal.prometheus.impl.InstanceMetaCollector;
-import com.alibaba.otter.canal.prometheus.impl.MemoryStoreCollector;
-import com.alibaba.otter.canal.prometheus.impl.PrometheusCanalEventDownStreamHandler;
-import com.alibaba.otter.canal.sink.CanalEventSink;
-import com.alibaba.otter.canal.sink.entry.EntryEventSink;
-import com.alibaba.otter.canal.store.CanalStoreException;
+import com.alibaba.otter.canal.prometheus.impl.*;
 import io.prometheus.client.Collector;
 import io.prometheus.client.CollectorRegistry;
 import org.slf4j.Logger;
@@ -20,79 +15,71 @@ import java.util.List;
  */
 public class CanalInstanceExports {
 
-    private static final Logger      logger         = LoggerFactory.getLogger(CanalInstanceExports.class);
+    private static final Logger      logger           = LoggerFactory.getLogger(CanalInstanceExports.class);
+    public static final String       DEST             = "destination";
+    public static final String[]     DEST_LABELS      = {DEST};
+    public static final List<String> DEST_LABELS_LIST = Collections.singletonList(DEST);
+    private final Collector          storeCollector;
+    private final Collector          entryCollector;
+    private final Collector          metaCollector;
+    private final Collector          sinkCollector;
+    private final Collector          parserCollector;
 
-    public static final String[]     labels         = {"destination"};
-
-    public static final List<String> labelList      = Collections.singletonList(labels[0]);
-
-    private final String             destination;
-
-    private Collector                storeCollector;
-
-    private Collector                delayCollector;
-
-    private Collector                metaCollector;
-
-    private CanalInstanceExports(CanalInstance instance) {
-        this.destination = instance.getDestination();
-        initDelayGauge(instance);
-        initStoreCollector(instance);
-        initMetaCollector(instance);
+    private CanalInstanceExports() {
+        this.storeCollector = StoreCollector.instance();
+        this.entryCollector = EntryCollector.instance();
+        this.metaCollector = MetaCollector.instance();
+        this.sinkCollector = SinkCollector.instance();
+        this.parserCollector = ParserCollector.instance();
     }
 
-
-
-    static CanalInstanceExports forInstance(CanalInstance instance) {
-        return new CanalInstanceExports(instance);
+    private static class SingletonHolder {
+        private static final CanalInstanceExports SINGLETON = new CanalInstanceExports();
     }
 
-    void register() {
-        if (delayCollector != null) {
-            delayCollector.register();
-        }
-        if (storeCollector != null) {
-            storeCollector.register();
-        }
-        if (metaCollector != null) {
-            metaCollector.register();
-        }
+    public static CanalInstanceExports instance() {
+        return SingletonHolder.SINGLETON;
     }
 
-    void unregister() {
-        if (delayCollector != null) {
-            CollectorRegistry.defaultRegistry.unregister(delayCollector);
-        }
-        if (storeCollector != null) {
-            CollectorRegistry.defaultRegistry.unregister(storeCollector);
-        }
-        if (metaCollector != null) {
-            CollectorRegistry.defaultRegistry.unregister(metaCollector);
-        }
+    public void initialize() {
+        storeCollector.register();
+        entryCollector.register();
+        metaCollector.register();
+        sinkCollector.register();
+        parserCollector.register();
     }
 
-    private void initDelayGauge(CanalInstance instance) {
-        CanalEventSink sink = instance.getEventSink();
-        if (sink instanceof EntryEventSink) {
-            EntryEventSink entryEventSink = (EntryEventSink) sink;
-            // TODO ensure not to add handler again
-            PrometheusCanalEventDownStreamHandler handler = new PrometheusCanalEventDownStreamHandler(destination);
-            entryEventSink.addHandler(handler);
-            delayCollector = handler.getCollector();
-        } else {
-            logger.warn("This impl register metrics for only EntryEventSink, skip.");
-        }
+    public void terminate() {
+        CollectorRegistry.defaultRegistry.unregister(storeCollector);
+        CollectorRegistry.defaultRegistry.unregister(entryCollector);
+        CollectorRegistry.defaultRegistry.unregister(metaCollector);
+        CollectorRegistry.defaultRegistry.unregister(sinkCollector);
+        CollectorRegistry.defaultRegistry.unregister(parserCollector);
     }
 
-    private void initStoreCollector(CanalInstance instance) {
-        try {
-            storeCollector = new MemoryStoreCollector(instance.getEventStore(), destination);
-        } catch (CanalStoreException cse) {
-            logger.warn("Failed to register metrics for destination {}.", destination, cse);
-        }
+    void register(CanalInstance instance) {
+        requiredInstanceRegistry(storeCollector).register(instance);
+        requiredInstanceRegistry(entryCollector).register(instance);
+        requiredInstanceRegistry(metaCollector).register(instance);
+        requiredInstanceRegistry(sinkCollector).register(instance);
+        requiredInstanceRegistry(parserCollector).register(instance);
+        logger.info("Successfully register metrics for instance {}.", instance.getDestination());
     }
 
-    private void initMetaCollector(CanalInstance instance) {
-        metaCollector = new InstanceMetaCollector(instance);
+    void unregister(CanalInstance instance) {
+        requiredInstanceRegistry(storeCollector).unregister(instance);
+        requiredInstanceRegistry(entryCollector).unregister(instance);
+        requiredInstanceRegistry(metaCollector).unregister(instance);
+        requiredInstanceRegistry(sinkCollector).unregister(instance);
+        requiredInstanceRegistry(parserCollector).unregister(instance);
+        logger.info("Successfully unregister metrics for instance {}.", instance.getDestination());
     }
+
+    private InstanceRegistry requiredInstanceRegistry(Collector collector) {
+        if (!(collector instanceof InstanceRegistry)) {
+            throw new IllegalArgumentException("Canal prometheus collector need to implement InstanceRegistry.");
+        }
+        return (InstanceRegistry) collector;
+    }
+
 }

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalServerExports.java

@@ -1,21 +0,0 @@
-package com.alibaba.otter.canal.prometheus;
-
-import com.alibaba.otter.canal.prometheus.impl.InboundThroughputAspect;
-import com.alibaba.otter.canal.prometheus.impl.OutboundThroughputAspect;
-
-/**
- * @author Chuanyi Li
- */
-public class CanalServerExports {
-
-    private static boolean initialized = false;
-
-    public static synchronized void initialize() {
-        if (!initialized) {
-            InboundThroughputAspect.getCollector().register();
-            OutboundThroughputAspect.getCollector().register();
-            initialized = true;
-        }
-    }
-
-}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/InstanceRegistry.java

@@ -0,0 +1,14 @@
+package com.alibaba.otter.canal.prometheus;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+
+/**
+ * @author Chuanyi Li
+ */
+public interface InstanceRegistry {
+
+    void register(CanalInstance instance);
+
+    void unregister(CanalInstance instance);
+
+}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java

@@ -1,6 +1,8 @@
 package com.alibaba.otter.canal.prometheus;
 
 import com.alibaba.otter.canal.instance.core.CanalInstance;
+import com.alibaba.otter.canal.prometheus.impl.PrometheusClientInstanceProfiler;
+import com.alibaba.otter.canal.server.netty.ClientInstanceProfiler;
 import com.alibaba.otter.canal.spi.CanalMetricsService;
 import io.prometheus.client.exporter.HTTPServer;
 import io.prometheus.client.hotspot.DefaultExports;
@@ -8,23 +10,24 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+
+import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler.NOP;
+import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler.profiler;
 
 /**
  * @author Chuanyi Li
  */
 public class PrometheusService implements CanalMetricsService {
 
-    private static final Logger                           logger  = LoggerFactory.getLogger(PrometheusService.class);
-
-    private final Map<String, CanalInstanceExports>       exports = new ConcurrentHashMap<String, CanalInstanceExports>();
-
-    private volatile boolean                              running = false;
-
-    private HTTPServer                                    server;
+    private static final Logger          logger          = LoggerFactory.getLogger(PrometheusService.class);
+    private final CanalInstanceExports   instanceExports;
+    private volatile boolean             running         = false;
+    private HTTPServer                   server;
+    private final ClientInstanceProfiler clientProfiler;
 
     private PrometheusService() {
+        this.instanceExports = CanalInstanceExports.instance();
+        this.clientProfiler = PrometheusClientInstanceProfiler.instance();
     }
 
     private static class SingletonHolder {
@@ -48,8 +51,11 @@ public class PrometheusService implements CanalMetricsService {
         try {
             // JVM exports
             DefaultExports.initialize();
-            // Canal server level exports
-            CanalServerExports.initialize();
+            instanceExports.initialize();
+            if (!clientProfiler.isStart()) {
+                clientProfiler.start();
+            }
+            profiler().setInstanceProfiler(clientProfiler);
         } catch (Throwable t) {
             logger.warn("Unable to initialize server exports.", t);
         }
@@ -60,14 +66,17 @@ public class PrometheusService implements CanalMetricsService {
     @Override
     public void terminate() {
         running = false;
-        // Normally, service should be terminated at canal shutdown.
-        // No need to unregister instance exports explicitly.
-        // But for the sake of safety, unregister them.
-        for (CanalInstanceExports ie : exports.values()) {
-            ie.unregister();
-        }
-        if (server != null) {
-            server.stop();
+        try {
+            instanceExports.terminate();
+            if (clientProfiler.isStart()) {
+                clientProfiler.stop();
+            }
+            profiler().setInstanceProfiler(NOP);
+            if (server != null) {
+                server.stop();
+            }
+        } catch (Throwable t) {
+            logger.warn("Something happened while terminating.", t);
         }
     }
 
@@ -83,9 +92,7 @@ public class PrometheusService implements CanalMetricsService {
             return;
         }
         try {
-            CanalInstanceExports export = CanalInstanceExports.forInstance(instance);
-            export.register();
-            exports.put(instance.getDestination(), export);
+            instanceExports.register(instance);
         } catch (Throwable t) {
             logger.warn("Unable to register instance exports for {}.", instance.getDestination(), t);
         }
@@ -98,13 +105,11 @@ public class PrometheusService implements CanalMetricsService {
             logger.warn("Try unregister metrics after destination {} is stopped.", instance.getDestination());
         }
         try {
-            CanalInstanceExports export = exports.remove(instance.getDestination());
-            if (export != null) {
-                export.unregister();
-            }
+            instanceExports.unregister(instance);
         } catch (Throwable t) {
             logger.warn("Unable to unregister instance exports for {}.", instance.getDestination(), t);
         }
         logger.info("Unregister metrics for destination {}.", instance.getDestination());
     }
+
 }

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java

@@ -0,0 +1,153 @@
+package com.alibaba.otter.canal.prometheus.impl;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+import com.alibaba.otter.canal.prometheus.InstanceRegistry;
+import com.alibaba.otter.canal.sink.CanalEventDownStreamHandler;
+import com.alibaba.otter.canal.sink.CanalEventSink;
+import com.alibaba.otter.canal.sink.entry.EntryEventSink;
+import com.google.common.base.Preconditions;
+import io.prometheus.client.Collector;
+import io.prometheus.client.Counter;
+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.Collections;
+import java.util.List;
+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_LABELS_LIST;
+
+/**
+ * @author Chuanyi Li
+ */
+public class EntryCollector extends Collector implements InstanceRegistry {
+
+    private static final Logger                             logger            = LoggerFactory.getLogger(SinkCollector.class);
+    private static final String                             DELAY             = "canal_instance_traffic_delay";
+    private static final String                             TRANSACTION       = "canal_instance_transactions";
+    private static final String                             ROW_EVENTS        = "canal_instance_row_events";
+    private static final String                             ROWS_COUNTER      = "canal_instance_rows_counter";
+    private static final String                             DELAY_HELP        = "Traffic delay of canal instance in milliseconds";
+    private static final String                             TRANSACTION_HELP  = "Transactions counter of canal instance";
+    private static final String                             ROW_EVENTS_HELP   = "Rowdata events counter of canal instance";
+    private static final String                             ROWS_COUNTER_HELP = "Rows counter of canal instance";
+    private final ConcurrentMap<String, EntryMetricsHolder> instances        = new ConcurrentHashMap<String, EntryMetricsHolder>();
+
+    private EntryCollector() {}
+
+    private static class SingletonHolder {
+        private static final EntryCollector SINGLETON = new EntryCollector();
+    }
+
+    public static EntryCollector instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        GaugeMetricFamily delay = new GaugeMetricFamily(DELAY,
+                DELAY_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily transactions = new CounterMetricFamily(TRANSACTION,
+                TRANSACTION_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily rowEvents = new CounterMetricFamily(ROW_EVENTS,
+                ROW_EVENTS_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily rowsCounter = new CounterMetricFamily(ROWS_COUNTER,
+                ROWS_COUNTER_HELP, DEST_LABELS_LIST);
+        for (EntryMetricsHolder emh : instances.values()) {
+            long now = System.currentTimeMillis();
+            long latest = emh.latestExecTime.get();
+            if (now >= latest) {
+                delay.addMetric(emh.destLabelValues, (now - latest));
+            }
+            transactions.addMetric(emh.destLabelValues, emh.transactionCounter.doubleValue());
+            rowEvents.addMetric(emh.destLabelValues, emh.rowEventCounter.doubleValue());
+            rowsCounter.addMetric(emh.destLabelValues, emh.rowsCounter.doubleValue());
+        }
+        mfs.add(delay);
+        mfs.add(transactions);
+        mfs.add(rowEvents);
+        mfs.add(rowsCounter);
+        return mfs;
+    }
+
+    @Override
+    public void register(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        EntryMetricsHolder holder = new EntryMetricsHolder();
+        holder.destLabelValues = Collections.singletonList(destination);
+        CanalEventSink sink = instance.getEventSink();
+        if (!(sink instanceof EntryEventSink)) {
+            throw new IllegalArgumentException("CanalEventSink must be EntryEventSink");
+        }
+        EntryEventSink entrySink = (EntryEventSink) sink;
+        PrometheusCanalEventDownStreamHandler handler = assembleHandler(entrySink);
+        holder.latestExecTime = handler.getLatestExecuteTime();
+        holder.transactionCounter = handler.getTransactionCounter();
+        holder.rowEventCounter = handler.getRowEventCounter();
+        holder.rowsCounter = handler.getRowsCounter();
+        Preconditions.checkNotNull(holder.latestExecTime);
+        Preconditions.checkNotNull(holder.transactionCounter);
+        Preconditions.checkNotNull(holder.rowEventCounter);
+        Preconditions.checkNotNull(holder.rowsCounter);
+        EntryMetricsHolder old = instances.put(destination, holder);
+        if (old != null) {
+            logger.warn("Remove stale EntryCollector for instance {}.", destination);
+        }
+    }
+
+    @Override
+    public void unregister(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        CanalEventSink sink = instance.getEventSink();
+        if (!(sink instanceof EntryEventSink)) {
+            throw new IllegalArgumentException("CanalEventSink must be EntryEventSink");
+        }
+        unloadHandler((EntryEventSink) sink);
+        instances.remove(destination);
+    }
+
+    private PrometheusCanalEventDownStreamHandler assembleHandler(EntryEventSink entrySink) {
+        PrometheusCanalEventDownStreamHandler ph = new PrometheusCanalEventDownStreamHandler();
+        List<CanalEventDownStreamHandler> handlers = entrySink.getHandlers();
+        for (CanalEventDownStreamHandler handler : handlers) {
+            if (handler instanceof PrometheusCanalEventDownStreamHandler) {
+                throw new IllegalStateException("PrometheusCanalEventDownStreamHandler already exists in handlers.");
+            }
+        }
+        entrySink.addHandler(ph, 0);
+        return ph;
+    }
+
+    private void unloadHandler(EntryEventSink entrySink) {
+        List<CanalEventDownStreamHandler> handlers = entrySink.getHandlers();
+        int i = 0;
+        for (; i < handlers.size(); i++) {
+            if (handlers.get(i) instanceof PrometheusCanalEventDownStreamHandler) {
+                break;
+            }
+        }
+        entrySink.removeHandler(i);
+        // Ensure no PrometheusCanalEventDownStreamHandler
+        handlers = entrySink.getHandlers();
+        for (CanalEventDownStreamHandler handler : handlers) {
+            if (handler instanceof PrometheusCanalEventDownStreamHandler) {
+                throw new IllegalStateException("Multiple prometheusCanalEventDownStreamHandler exists in handlers.");
+            }
+        }
+    }
+
+    private class EntryMetricsHolder {
+        private AtomicLong   latestExecTime;
+        private AtomicLong   transactionCounter;
+        private AtomicLong   rowEventCounter;
+        private AtomicLong   rowsCounter;
+        private List<String> destLabelValues;
+    }
+
+}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InboundThroughputAspect.java

@@ -1,79 +0,0 @@
-package com.alibaba.otter.canal.prometheus.impl;
-
-import io.prometheus.client.Collector;
-import io.prometheus.client.CounterMetricFamily;
-import org.aspectj.lang.annotation.After;
-import org.aspectj.lang.annotation.Aspect;
-import org.aspectj.lang.annotation.Pointcut;
-import org.jctools.maps.ConcurrentAutoTable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * @author Chuanyi Li
- */
-@Aspect
-public class InboundThroughputAspect {
-
-    private static final Logger              logger    = LoggerFactory.getLogger(InboundThroughputAspect.class);
-
-    /**
-     *  Support highly scalable counters
-     *  @see ConcurrentAutoTable
-     */
-    private static final ConcurrentAutoTable total     = new ConcurrentAutoTable();
-
-    private static final Collector           collector = new InboundThroughputCollector();
-
-    public static Collector getCollector() {
-        return collector;
-    }
-
-    @Pointcut("call(byte[] com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel.read(..))")
-    public void read() {}
-
-    @Pointcut("call(void com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel.read(..)) ")
-    public void readBytes() {}
-
-    //nested read, just eliminate them.
-    @Pointcut("withincode(* com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel.read(..))")
-    public void nestedCall() {}
-
-    @After("read() && !nestedCall()  && args(len, ..)")
-    public void recordRead(int len) {
-        accumulateBytes(len);
-    }
-
-    @After("readBytes() && !nestedCall() && args(.., len, timeout)")
-    public void recordReadBytes(int len, int timeout) {
-        accumulateBytes(len);
-    }
-
-    private void accumulateBytes(int count) {
-        try {
-            total.add(count);
-        } catch (Throwable t) {
-            //Catch every Throwable, rather than break the business logic.
-            logger.warn("Error while accumulate inbound bytes.", t);
-        }
-    }
-
-    public static class InboundThroughputCollector extends Collector {
-
-        private InboundThroughputCollector() {}
-
-        @Override
-        public List<MetricFamilySamples> collect() {
-            List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
-            CounterMetricFamily bytes = new CounterMetricFamily("canal_net_inbound_bytes",
-                    "Total socket inbound bytes of canal server.",
-                    total.get());
-            mfs.add(bytes);
-            return mfs;
-        }
-    }
-
-}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InstanceMetaCollector.java

@@ -1,64 +0,0 @@
-package com.alibaba.otter.canal.prometheus.impl;
-
-import com.alibaba.otter.canal.instance.core.CanalInstance;
-import com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring;
-import com.alibaba.otter.canal.meta.CanalMetaManager;
-import com.alibaba.otter.canal.prometheus.CanalInstanceExports;
-import com.alibaba.otter.canal.protocol.ClientIdentity;
-import io.prometheus.client.Collector;
-import io.prometheus.client.GaugeMetricFamily;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-/**
- * @author Chuanyi Li
- */
-public class InstanceMetaCollector extends Collector {
-
-    private static final List<String> InfoLabel    = Arrays.asList("destination", "mode");
-
-    private CanalMetaManager          metaManager;
-
-    private final String              destination;
-
-    private final String              mode;
-
-    private final String              subsHelp;
-
-    public InstanceMetaCollector(CanalInstance instance) {
-        if (instance == null) {
-            throw new IllegalArgumentException("CanalInstance must not be null.");
-        }
-        if (instance instanceof CanalInstanceWithSpring) {
-            mode = "spring";
-        } else {
-            mode = "manager";
-        }
-        this.metaManager = instance.getMetaManager();
-        this.destination = instance.getDestination();
-        this.subsHelp = "Subscriptions of canal instance " + destination;
-    }
-
-    @Override
-    public List<MetricFamilySamples> collect() {
-        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
-        GaugeMetricFamily instanceInfo = new GaugeMetricFamily(
-                "canal_instance",
-                "Canal instance",
-                InfoLabel);
-        instanceInfo.addMetric(Arrays.asList(destination, mode), 1);
-        mfs.add(instanceInfo);
-        if (metaManager.isStart()) {
-            // client id = hardcode 1001, 目前没有意义
-            List<ClientIdentity> subs = metaManager.listAllSubscribeInfo(destination);
-            GaugeMetricFamily subscriptions = new GaugeMetricFamily(
-                    "canal_instance_subscription",
-                    subsHelp, CanalInstanceExports.labelList);
-            subscriptions.addMetric(Arrays.asList(destination), subs.size());
-            mfs.add(subscriptions);
-        }
-        return mfs;
-    }
-}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MemoryStoreCollector.java

@@ -1,75 +0,0 @@
-package com.alibaba.otter.canal.prometheus.impl;
-
-import com.alibaba.otter.canal.prometheus.CanalInstanceExports;
-import com.alibaba.otter.canal.store.CanalEventStore;
-import com.alibaba.otter.canal.store.CanalStoreException;
-import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
-import io.prometheus.client.Collector;
-import io.prometheus.client.CounterMetricFamily;
-
-import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * @author Chuanyi Li
- */
-public class MemoryStoreCollector extends Collector {
-
-    private static final Class<MemoryEventStoreWithBuffer> clazz  = MemoryEventStoreWithBuffer.class;
-
-    private final String                                   destination;
-
-    private final AtomicLong                               putSequence;
-
-    private final AtomicLong                               ackSequence;
-
-    private final String                                   putHelp;
-
-    private final String                                   ackHelp;
-
-    public MemoryStoreCollector(CanalEventStore store, String destination) {
-        this.destination = destination;
-        if (!(store instanceof MemoryEventStoreWithBuffer)) {
-            throw new IllegalArgumentException("EventStore must be MemoryEventStoreWithBuffer");
-        }
-        MemoryEventStoreWithBuffer ms = (MemoryEventStoreWithBuffer) store;
-        putSequence = getDeclaredValue(ms, "putSequence");
-        ackSequence = getDeclaredValue(ms, "ackSequence");
-        putHelp = "Produced sequence of canal instance " + destination;
-        ackHelp = "Consumed sequence of canal instance " + destination;
-    }
-
-    @Override
-    public List<MetricFamilySamples> collect() {
-        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
-        CounterMetricFamily put = new CounterMetricFamily("canal_instance_store_produce_seq",
-                putHelp, Arrays.asList(CanalInstanceExports.labels));
-        put.addMetric(Collections.singletonList(destination), putSequence.doubleValue());
-        mfs.add(put);
-        CounterMetricFamily ack = new CounterMetricFamily("canal_instance_store_consume_seq",
-                ackHelp, Arrays.asList(CanalInstanceExports.labels));
-        ack.addMetric(Collections.singletonList(destination), ackSequence.doubleValue());
-        mfs.add(ack);
-        return mfs;
-    }
-
-    @SuppressWarnings("unchecked")
-    private static <T> T getDeclaredValue(MemoryEventStoreWithBuffer store, String name) {
-        T value;
-        try {
-            Field putField = clazz.getDeclaredField(name);
-            putField.setAccessible(true);
-            value = (T) putField.get(store);
-        } catch (NoSuchFieldException e) {
-            throw new CanalStoreException(e);
-        } catch (IllegalAccessException e) {
-            throw new CanalStoreException(e);
-        }
-        return value;
-    }
-
-}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MetaCollector.java

@@ -0,0 +1,91 @@
+package com.alibaba.otter.canal.prometheus.impl;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+import com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring;
+import com.alibaba.otter.canal.meta.CanalMetaManager;
+import com.alibaba.otter.canal.prometheus.InstanceRegistry;
+import com.alibaba.otter.canal.protocol.ClientIdentity;
+import com.google.common.base.Preconditions;
+import io.prometheus.client.Collector;
+import io.prometheus.client.GaugeMetricFamily;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST;
+
+/**
+ * @author Chuanyi Li
+ */
+public class MetaCollector extends Collector implements InstanceRegistry {
+
+    private static final List<String>                      INFO_LABELS_LIST  = Arrays.asList("destination", "mode");
+    private static final Logger                            logger            = LoggerFactory.getLogger(MetaCollector.class);
+    private static final String                            INSTANCE          = "canal_instance";
+    private static final String                            INSTANCE_HELP     = "Canal instance";
+    private static final String                            SUBSCRIPTION      = "canal_instance_subscriptions";
+    private static final String                            SUBSCRIPTION_HELP = "Canal instance subscriptions";
+    private final ConcurrentMap<String, MetaMetricsHolder> instances         = new ConcurrentHashMap<String, MetaMetricsHolder>();
+
+    private MetaCollector() {}
+
+    private static class SingletonHolder {
+        private static final MetaCollector SINGLETON = new MetaCollector();
+    }
+
+    public static MetaCollector instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        GaugeMetricFamily instanceInfo = new GaugeMetricFamily(INSTANCE,
+                INSTANCE_HELP, INFO_LABELS_LIST);
+        GaugeMetricFamily subsInfo = new GaugeMetricFamily(SUBSCRIPTION,
+                SUBSCRIPTION_HELP, DEST_LABELS_LIST);
+        for (Map.Entry<String, MetaMetricsHolder> nme : instances.entrySet()) {
+            final String destination = nme.getKey();
+            final MetaMetricsHolder nmh = nme.getValue();
+            instanceInfo.addMetric(nmh.infoLabelValues, 1);
+            List<ClientIdentity> subs = nmh.metaManager.listAllSubscribeInfo(destination);
+            int count = subs == null ? 0 : subs.size();
+            subsInfo.addMetric(nmh.destLabelValues, count);
+        }
+        mfs.add(instanceInfo);
+        mfs.add(subsInfo);
+        return mfs;
+    }
+
+    @Override
+    public void register(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        MetaMetricsHolder holder = new MetaMetricsHolder();
+        String mode = (instance instanceof CanalInstanceWithSpring) ? "spring" : "manager";
+        holder.infoLabelValues = Arrays.asList(destination, mode);
+        holder.destLabelValues = Collections.singletonList(destination);
+        holder.metaManager = instance.getMetaManager();
+        Preconditions.checkNotNull(holder.metaManager);
+        MetaMetricsHolder old = instances.put(destination, holder);
+        if (old != null) {
+            logger.warn("Remove stale MetaCollector for instance {}.", destination);
+        }
+    }
+
+    @Override
+    public void unregister(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        instances.remove(destination);
+    }
+
+    private class MetaMetricsHolder {
+        private List<String>     infoLabelValues;
+        private List<String>     destLabelValues;
+        private CanalMetaManager metaManager;
+    }
+
+
+}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/OutboundThroughputAspect.java

@@ -1,80 +0,0 @@
-package com.alibaba.otter.canal.prometheus.impl;
-
-import io.prometheus.client.Collector;
-import io.prometheus.client.CounterMetricFamily;
-import org.aspectj.lang.annotation.After;
-import org.aspectj.lang.annotation.Aspect;
-import org.aspectj.lang.annotation.Pointcut;
-import org.jboss.netty.channel.Channel;
-import org.jctools.maps.ConcurrentAutoTable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
-import static com.alibaba.otter.canal.server.netty.NettyUtils.HEADER_LENGTH;
-
-/**
- * @author Chuanyi Li
- */
-@Aspect
-public class OutboundThroughputAspect {
-    private static final Logger              logger    = LoggerFactory.getLogger(OutboundThroughputAspect.class);
-
-    /**
-     *  Support highly scalable counters
-     *  @see ConcurrentAutoTable
-     */
-    private static final ConcurrentAutoTable total     = new ConcurrentAutoTable();
-
-    private static final Collector           collector = new OutboundThroughputCollector();
-
-    public static Collector getCollector() {
-        return collector;
-    }
-
-    @Pointcut("call(* com.alibaba.otter.canal.server.netty.NettyUtils.write(..))")
-    public void write() {}
-
-    //nested read, just eliminate them.
-    @Pointcut("withincode(* com.alibaba.otter.canal.server.netty.NettyUtils.write(..))")
-    public void nestedCall() {}
-
-    @After("write() && !nestedCall() && args(ch, bytes, ..)")
-    public void recordWriteBytes(Channel ch, byte[] bytes) {
-        if (bytes != null) {
-            accumulateBytes(HEADER_LENGTH + bytes.length);
-        }
-    }
-
-    @After("write() && !nestedCall() && args(ch, buf, ..)")
-    public void recordWriteBuffer(Channel ch, ByteBuffer buf) {
-        if (buf != null) {
-            total.add(HEADER_LENGTH + buf.limit());
-        }
-    }
-    private void accumulateBytes(int count) {
-        try {
-            total.add(count);
-        } catch (Throwable t) {
-            //Catch every Throwable, rather than break the business logic.
-            logger.warn("Error while accumulate inbound bytes.", t);
-        }
-    }
-
-    public static class OutboundThroughputCollector extends Collector {
-
-        private OutboundThroughputCollector() {}
-
-        @Override public List<MetricFamilySamples> collect() {
-            List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
-            CounterMetricFamily bytes = new CounterMetricFamily("canal_net_outbound_bytes",
-                    "Total socket outbound bytes of canal server.",
-                    total.get());
-            mfs.add(bytes);
-            return mfs;
-        }
-    }
-}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java

@@ -0,0 +1,116 @@
+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 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;
+import java.util.Collections;
+import java.util.List;
+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;
+
+/**
+ * @author Chuanyi Li
+ */
+public class ParserCollector extends Collector implements InstanceRegistry {
+
+    private static final Logger                              logger                = LoggerFactory.getLogger(ParserCollector.class);
+    private static final long                                NANO_PER_MILLI        = 1000 * 1000L;
+    private static final String                              PUBLISH_BLOCKING      = "canal_instance_publish_blocking_time";
+    private static final String                              RECEIVED_BINLOG       = "canal_instance_received_binlog_bytes";
+    private static final String                              PARSER_MODE           = "canal_instance_parser_mode";
+    private static final String                              MODE_LABEL            = "parallel";
+    private static final String                              PUBLISH_BLOCKING_HELP = "Publish blocking time of dump thread in milliseconds";
+    private static final String                              RECEIVED_BINLOG_HELP  = "Received binlog bytes";
+    private static final String                              MODE_HELP             = "Parser mode(parallel/serial) of instance";
+    private final List<String>                               modeLabels            = Arrays.asList(DEST, MODE_LABEL);
+    private final ConcurrentMap<String, ParserMetricsHolder> instances             = new ConcurrentHashMap<String, ParserMetricsHolder>();
+
+    private ParserCollector() {}
+
+    private static class SingletonHolder {
+        private static final ParserCollector SINGLETON = new ParserCollector();
+    }
+
+    public static ParserCollector instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        boolean hasParallel = false;
+        CounterMetricFamily bytesCounter = new CounterMetricFamily(RECEIVED_BINLOG,
+                RECEIVED_BINLOG_HELP, DEST_LABELS_LIST);
+        GaugeMetricFamily modeGauge = new GaugeMetricFamily(PARSER_MODE,
+                MODE_HELP, modeLabels);
+        CounterMetricFamily blockingCounter = new CounterMetricFamily(PUBLISH_BLOCKING,
+                PUBLISH_BLOCKING_HELP, DEST_LABELS_LIST);
+        for (ParserMetricsHolder emh : instances.values()) {
+            if (emh.isParallel) {
+                blockingCounter.addMetric(emh.destLabelValues, (emh.eventsPublishBlockingTime.doubleValue() / NANO_PER_MILLI));
+                hasParallel = true;
+            }
+            modeGauge.addMetric(emh.modeLabelValues, 1);
+            bytesCounter.addMetric(emh.destLabelValues, emh.receivedBinlogBytes.doubleValue());
+
+        }
+        mfs.add(bytesCounter);
+        mfs.add(modeGauge);
+        if (hasParallel) {
+            mfs.add(blockingCounter);
+        }
+        return mfs;
+    }
+
+    @Override
+    public void register(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        ParserMetricsHolder holder = new ParserMetricsHolder();
+        CanalEventParser parser = instance.getEventParser();
+        if (!(parser instanceof MysqlEventParser)) {
+            throw new IllegalArgumentException("CanalEventParser must be MysqlEventParser");
+        }
+        MysqlEventParser mysqlParser = (MysqlEventParser) parser;
+        holder.destLabelValues = Collections.singletonList(destination);
+        holder.modeLabelValues = Arrays.asList(destination, Boolean.toString(mysqlParser.isParallel()));
+        holder.eventsPublishBlockingTime = mysqlParser.getEventsPublishBlockingTime();
+        holder.receivedBinlogBytes = mysqlParser.getReceivedBinlogBytes();
+        holder.isParallel = mysqlParser.isParallel();
+        Preconditions.checkNotNull(holder.eventsPublishBlockingTime);
+        Preconditions.checkNotNull(holder.receivedBinlogBytes);
+        ParserMetricsHolder old = instances.put(destination, holder);
+        if (old != null) {
+            logger.warn("Remove stale ParserCollector for instance {}.", destination);
+        }
+    }
+
+    @Override
+    public void unregister(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        instances.remove(destination);
+    }
+
+    private class ParserMetricsHolder {
+        private List<String> destLabelValues;
+        private List<String> modeLabelValues;
+        private AtomicLong   receivedBinlogBytes;
+        private AtomicLong   eventsPublishBlockingTime;
+        private boolean      isParallel;
+    }
+
+}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java

@@ -1,73 +1,90 @@
 package com.alibaba.otter.canal.prometheus.impl;
 
-import com.alibaba.otter.canal.prometheus.CanalInstanceExports;
+import com.alibaba.otter.canal.protocol.CanalEntry;
+import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
 import com.alibaba.otter.canal.sink.AbstractCanalEventDownStreamHandler;
 import com.alibaba.otter.canal.store.model.Event;
-import io.prometheus.client.Collector;
-import io.prometheus.client.GaugeMetricFamily;
 
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * @author Chuanyi Li
  */
 public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDownStreamHandler<List<Event>> {
 
-    private final Collector     collector;
-
-    private long                latestExecuteTime = 0L;
-
-    private static final String DELAY_NAME        = "canal_instance_traffic_delay";
-
-    private final String        delayHelpName;
-
-    private final List<String>  labelValues;
-
-    public PrometheusCanalEventDownStreamHandler(final String destination) {
-        this.delayHelpName = "Traffic delay of canal instance " + destination + " in seconds.";
-        this.labelValues = Collections.singletonList(destination);
-        collector = new Collector() {
-            @Override
-            public List<MetricFamilySamples> collect() {
-                List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
-                long now = System.currentTimeMillis();
-                GaugeMetricFamily delay = new GaugeMetricFamily(
-                        DELAY_NAME,
-                        delayHelpName,
-                        CanalInstanceExports.labelList);
-                double d = 0.0;
-                if (latestExecuteTime > 0) {
-                    d = now - latestExecuteTime;
-                }
-                d = d > 0.0 ? (d / 1000) : 0.0;
-                delay.addMetric(labelValues, d);
-                mfs.add(delay);
-                return mfs;
-            }
-        };
-    }
+    private final AtomicLong latestExecuteTime  = new AtomicLong(0L);
+    private final AtomicLong transactionCounter = new AtomicLong(0L);
+    private final AtomicLong rowEventCounter    = new AtomicLong(0L);
+    private final AtomicLong rowsCounter        = new AtomicLong(0L);
 
     @Override
     public List<Event> before(List<Event> events) {
-        // TODO utilize MySQL master heartbeat packet to refresh delay if always no more events coming
-        // see: https://dev.mysql.com/worklog/task/?id=342
-        // heartbeats are sent by the master only if there is no
-        // more unsent events in the actual binlog file for a period longer that
-        // master_heartbeat_period.
+        long localExecTime = 0L;
         if (events != null && !events.isEmpty()) {
-            Event last = events.get(events.size() - 1);
-            long ts = last.getExecuteTime();
-            if (ts > latestExecuteTime) {
-                latestExecuteTime = ts;
+            for (Event e : events) {
+                EntryType type = e.getEntryType();
+                if (type == null) continue;
+                switch (type) {
+                    case TRANSACTIONBEGIN: {
+                        long exec = e.getExecuteTime();
+                        if (exec > 0) localExecTime = exec;
+                        break;
+                    }
+                    case ROWDATA: {
+                        long exec = e.getExecuteTime();
+                        if (exec > 0) localExecTime = exec;
+                        rowEventCounter.incrementAndGet();
+                        rowsCounter.addAndGet(e.getRowsCount());
+                        break;
+                    }
+                    case TRANSACTIONEND: {
+                        long exec = e.getExecuteTime();
+                        if (exec > 0) localExecTime = exec;
+                        transactionCounter.incrementAndGet();
+                        break;
+                    }
+                    case HEARTBEAT:
+                        CanalEntry.EventType eventType = e.getEventType();
+                        if (eventType == CanalEntry.EventType.MHEARTBEAT) {
+                            localExecTime = System.currentTimeMillis();
+                        }
+                        break;
+                    default:
+                        break;
+                }
+            }
+            if (localExecTime > 0) {
+                latestExecuteTime.lazySet(localExecTime);
             }
         }
         return events;
     }
 
-    public Collector getCollector() {
-        return this.collector;
+    @Override
+    public void start() {
+
+        super.start();
     }
 
+    @Override
+    public void stop() {
+        super.stop();
+    }
+
+    public AtomicLong getLatestExecuteTime() {
+        return latestExecuteTime;
+    }
+
+    public AtomicLong getTransactionCounter() {
+        return transactionCounter;
+    }
+
+    public AtomicLong getRowsCounter() {
+        return rowsCounter;
+    }
+
+    public AtomicLong getRowEventCounter() {
+        return rowEventCounter;
+    }
 }

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfiler.java

@@ -0,0 +1,138 @@
+package com.alibaba.otter.canal.prometheus.impl;
+
+import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
+import com.alibaba.otter.canal.server.netty.ClientInstanceProfiler;
+import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator.ClientRequestResult;
+import io.prometheus.client.CollectorRegistry;
+import io.prometheus.client.Counter;
+import io.prometheus.client.Histogram;
+
+import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST;
+import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS;
+
+/**
+ * @author Chuanyi Li
+ */
+public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler {
+
+    private static final long   NANO_PER_MILLI = 1000 * 1000L;
+    private static final String PACKET_TYPE    = "canal_instance_client_packets";
+    private static final String OUTBOUND_BYTES = "canal_instance_client_bytes";
+    private static final String EMPTY_BATCHES  = "canal_instance_client_empty_batches";
+    private static final String ERRORS         = "canal_instance_client_request_error";
+    private static final String LATENCY        = "canal_instance_client_request_latency";
+    private final Counter       outboundCounter;
+    private final Counter       packetsCounter;
+    private final Counter       emptyBatchesCounter;
+    private final Counter       errorsCounter;
+    private final Histogram     responseLatency;
+    private volatile boolean    running        = false;
+
+    private static class SingletonHolder {
+        private static final PrometheusClientInstanceProfiler SINGLETON = new PrometheusClientInstanceProfiler();
+    }
+
+    public static PrometheusClientInstanceProfiler instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    private PrometheusClientInstanceProfiler() {
+        this.outboundCounter = Counter.build()
+                .labelNames(DEST_LABELS)
+                .name(OUTBOUND_BYTES)
+                .help("Total bytes sent to client.")
+                .create();
+        this.packetsCounter = Counter.build()
+                .labelNames(new String[]{DEST, "packetType"})
+                .name(PACKET_TYPE)
+                .help("Total packets sent to client.")
+                .create();
+        this.emptyBatchesCounter = Counter.build()
+                .labelNames(DEST_LABELS)
+                .name(EMPTY_BATCHES)
+                .help("Total empty batches sent to client.")
+                .create();
+        this.errorsCounter = Counter.build()
+                .labelNames(new String[]{DEST, "errorCode"})
+                .name(ERRORS)
+                .help("Total client request errors.")
+                .create();
+        this.responseLatency = Histogram.build()
+                .labelNames(DEST_LABELS)
+                .name(LATENCY)
+                .help("Client request latency.")
+                // buckets in milliseconds
+                .buckets(2.5, 10.0, 25.0, 100.0)
+                .create();
+    }
+
+    @Override
+    public void profiling(ClientRequestResult result) {
+        String destination = result.getDestination();
+        PacketType type = result.getType();
+        outboundCounter.labels(destination).inc(result.getAmount());
+        packetsCounter.labels(destination, type.name()).inc();
+        short errorCode = result.getErrorCode();
+        if (errorCode > 0) {
+            errorsCounter.labels(destination, Short.toString(errorCode)).inc();
+        }
+        long latency = result.getLatency();
+        responseLatency.labels(destination).observe(((double) latency) / NANO_PER_MILLI);
+        switch (type) {
+            case GET:
+                boolean empty = result.getEmpty();
+                if (empty) {
+                    emptyBatchesCounter.labels(destination).inc();
+                }
+                break;
+            // reserve for others
+            default:
+                break;
+        }
+    }
+
+    @Override
+    public void start() {
+        if (outboundCounter != null) {
+            outboundCounter.register();
+        }
+        if (packetsCounter != null) {
+            packetsCounter.register();
+        }
+        if (emptyBatchesCounter != null) {
+            emptyBatchesCounter.register();
+        }
+        if (errorsCounter != null) {
+            errorsCounter.register();
+        }
+        if (responseLatency != null) {
+            responseLatency.register();
+        }
+        running = true;
+    }
+
+    @Override
+    public void stop() {
+        running = false;
+        if (outboundCounter != null) {
+            CollectorRegistry.defaultRegistry.unregister(outboundCounter);
+        }
+        if (packetsCounter != null) {
+            CollectorRegistry.defaultRegistry.unregister(packetsCounter);
+        }
+        if (emptyBatchesCounter != null) {
+            CollectorRegistry.defaultRegistry.unregister(emptyBatchesCounter);
+        }
+        if (errorsCounter != null) {
+            CollectorRegistry.defaultRegistry.unregister(errorsCounter);
+        }
+        if (responseLatency != null) {
+            CollectorRegistry.defaultRegistry.unregister(responseLatency);
+        }
+    }
+
+    @Override
+    public boolean isStart() {
+        return running;
+    }
+}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java

@@ -0,0 +1,84 @@
+package com.alibaba.otter.canal.prometheus.impl;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+import com.alibaba.otter.canal.prometheus.InstanceRegistry;
+import com.alibaba.otter.canal.sink.CanalEventSink;
+import com.alibaba.otter.canal.sink.entry.EntryEventSink;
+import com.google.common.base.Preconditions;
+import io.prometheus.client.Collector;
+import io.prometheus.client.CounterMetricFamily;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+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_LABELS_LIST;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+/**
+ * @author Chuanyi Li
+ */
+public class SinkCollector extends Collector implements InstanceRegistry {
+
+    private static final Logger                            logger               = LoggerFactory.getLogger(SinkCollector.class);
+    private static final long                              NANO_PER_MILLI       = 1000 * 1000L;
+    private static final String                            SINK_BLOCKING_TIME   = "canal_instance_sink_blocking_time";
+    private static final String                            SINK_BLOCK_TIME_HELP = "Total sink blocking time in milliseconds";
+    private final ConcurrentMap<String, SinkMetricsHolder> instances            = new ConcurrentHashMap<String, SinkMetricsHolder>();
+
+    private SinkCollector() {}
+
+    private static class SingletonHolder {
+        private static final SinkCollector SINGLETON = new SinkCollector();
+    }
+
+    public static SinkCollector instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME,
+                SINK_BLOCK_TIME_HELP, DEST_LABELS_LIST);
+        for (SinkMetricsHolder smh : instances.values()) {
+            blockingCounter.addMetric(smh.destLabelValues, (smh.eventsSinkBlockingTime.doubleValue() / NANO_PER_MILLI));
+        }
+        mfs.add(blockingCounter);
+        return mfs;
+    }
+
+    @Override
+    public void register(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        SinkMetricsHolder holder = new SinkMetricsHolder();
+        holder.destLabelValues = Collections.singletonList(destination);
+        CanalEventSink sink = instance.getEventSink();
+        if (!(sink instanceof EntryEventSink)) {
+            throw new IllegalArgumentException("CanalEventSink must be EntryEventSink");
+        }
+        EntryEventSink entrySink = (EntryEventSink) sink;
+        holder.eventsSinkBlockingTime = entrySink.getEventsSinkBlockingTime();
+        Preconditions.checkNotNull(holder.eventsSinkBlockingTime);
+        SinkMetricsHolder old = instances.put(destination, holder);
+        if (old != null) {
+            logger.warn("Remote stale SinkCollector for instance {}.", destination);
+        }
+    }
+
+    @Override
+    public void unregister(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        instances.remove(destination);
+    }
+
+    private class SinkMetricsHolder {
+        private AtomicLong   eventsSinkBlockingTime;
+        private List<String> destLabelValues;
+    }
+}

prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java

@@ -0,0 +1,134 @@
+package com.alibaba.otter.canal.prometheus.impl;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+import com.alibaba.otter.canal.prometheus.InstanceRegistry;
+import com.alibaba.otter.canal.store.CanalEventStore;
+import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
+import com.alibaba.otter.canal.store.model.BatchMode;
+import com.google.common.base.Preconditions;
+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;
+import java.util.Collections;
+import java.util.List;
+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;
+
+/**
+ * @author Chuanyi Li
+ */
+public class StoreCollector extends Collector implements InstanceRegistry {
+
+    private static final Logger                             logger           = LoggerFactory.getLogger(SinkCollector.class);
+    private static final String                             PRODUCE          = "canal_instance_store_produce_seq";
+    private static final String                             CONSUME          = "canal_instance_store_consume_seq";
+    private static final String                             STORE            = "canal_instance_store";
+    private static final String                             PRODUCE_MEM      = "canal_instance_store_produce_mem";
+    private static final String                             CONSUME_MEM      = "canal_instance_store_consume_mem";
+    private static final String                             PRODUCE_HELP     = "Produced events counter of canal instance";
+    private static final String                             CONSUME_HELP     = "Consumed events counter of canal instance";
+    private static final String                             STORE_HELP       = "Canal instance info";
+    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<String, StoreMetricsHolder> instances        = new ConcurrentHashMap<String, StoreMetricsHolder>();
+    private final List<String>                              storeLabelsList  = Arrays.asList(DEST, "batchMode");
+
+    private StoreCollector() {}
+
+    private static class SingletonHolder {
+        private static final StoreCollector SINGLETON = new StoreCollector();
+    }
+
+    public static StoreCollector instance() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        CounterMetricFamily put = new CounterMetricFamily(PRODUCE,
+                PRODUCE_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily ack = new CounterMetricFamily(CONSUME,
+                CONSUME_HELP, DEST_LABELS_LIST);
+        GaugeMetricFamily store = new GaugeMetricFamily(STORE,
+                STORE_HELP, storeLabelsList);
+        CounterMetricFamily putMem = new CounterMetricFamily(PRODUCE_MEM,
+                PRODUCE_MEM_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily ackMem = new CounterMetricFamily(CONSUME_MEM,
+                CONSUME_MEM_HELP, DEST_LABELS_LIST);
+        boolean hasMem = false;
+        for (StoreMetricsHolder smh : instances.values()) {
+            final boolean isMem = smh.batchMode.isMemSize();
+            put.addMetric(smh.destLabelValues, smh.putSeq.doubleValue());
+            ack.addMetric(smh.destLabelValues, smh.ackSeq.doubleValue());
+            store.addMetric(smh.storeLabelValues, 1);
+            if (isMem) {
+                hasMem = true;
+                putMem.addMetric(smh.destLabelValues, smh.putMemSize.doubleValue());
+                ackMem.addMetric(smh.destLabelValues, smh.ackMemSize.doubleValue());
+            }
+        }
+        mfs.add(put);
+        mfs.add(ack);
+        mfs.add(store);
+        if (hasMem) {
+            mfs.add(putMem);
+            mfs.add(ackMem);
+        }
+        return mfs;
+    }
+
+    @Override
+    public void register(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        StoreMetricsHolder holder = new StoreMetricsHolder();
+        CanalEventStore store = instance.getEventStore();
+        if (!(store instanceof MemoryEventStoreWithBuffer)) {
+            throw new IllegalArgumentException("EventStore must be MemoryEventStoreWithBuffer");
+        }
+        MemoryEventStoreWithBuffer memStore = (MemoryEventStoreWithBuffer) store;
+        holder.batchMode = memStore.getBatchMode();
+        holder.putSeq = memStore.getPutSequence();
+        holder.ackSeq = memStore.getAckSequence();
+        holder.destLabelValues = Collections.singletonList(destination);
+        holder.storeLabelValues = Arrays.asList(destination, memStore.getBatchMode().name());
+        Preconditions.checkNotNull(holder.batchMode);
+        Preconditions.checkNotNull(holder.putSeq);
+        Preconditions.checkNotNull(holder.ackSeq);
+        if (holder.batchMode.isMemSize()) {
+            holder.putMemSize = memStore.getPutMemSize();
+            holder.ackMemSize = memStore.getAckMemSize();
+            Preconditions.checkNotNull(holder.putMemSize);
+            Preconditions.checkNotNull(holder.ackMemSize);
+        }
+        StoreMetricsHolder old = instances.putIfAbsent(destination, holder);
+        if (old != null) {
+            logger.warn("Remote stale StoreCollector for instance {}.", destination);
+        }
+    }
+
+    @Override
+    public void unregister(CanalInstance instance) {
+        final String destination = instance.getDestination();
+        instances.remove(destination);
+    }
+
+    private class StoreMetricsHolder {
+        private AtomicLong   putSeq;
+        private AtomicLong   ackSeq;
+        private BatchMode    batchMode;
+        private AtomicLong   putMemSize;
+        private AtomicLong   ackMemSize;
+        private List<String> destLabelValues;
+        private List<String> storeLabelValues;
+    }
+}

prometheus/src/main/resources/META-INF/aop.xml

@@ -1,11 +0,0 @@
-<aspectj>
-
-    <aspects>
-        <aspect name="com.alibaba.otter.canal.prometheus.impl.InboundThroughputAspect"/>
-        <aspect name="com.alibaba.otter.canal.prometheus.impl.OutboundThroughputAspect"/>
-    </aspects>
-    <weaver options="-verbose -showWeaveInfo">
-        <include within="com.alibaba.otter.canal..*"/>
-    </weaver>
-
-</aspectj>

protocol/src/main/java/com/alibaba/otter/canal/protocol/EntryProtocol.proto

@@ -194,6 +194,8 @@ enum EventType {
     /** XA **/
     XACOMMIT    =       13;
     XAROLLBACK  =		14;
+    /** MASTER HEARTBEAT **/
+    MHEARTBEAT  =       15;
 }
 
 /**数据库类型**/
@@ -201,4 +203,4 @@ enum Type {
     ORACLE		= 		1;
     MYSQL		= 		2;
     PGSQL		= 		3;
-}
+}

server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNettyProfiler.java

@@ -0,0 +1,39 @@
+package com.alibaba.otter.canal.server.netty;
+
+import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
+import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator.ClientRequestResult;
+
+/**
+ * @author Chuanyi Li
+ */
+public class CanalServerWithNettyProfiler {
+
+    public static final ClientInstanceProfiler NOP               = new DefaultClientInstanceProfiler();
+    private ClientInstanceProfiler             instanceProfiler;
+
+    private static class SingletonHolder {
+        private static CanalServerWithNettyProfiler SINGLETON = new CanalServerWithNettyProfiler();
+    }
+
+    private CanalServerWithNettyProfiler() {
+        this.instanceProfiler = NOP;
+    }
+
+    public static CanalServerWithNettyProfiler profiler() {
+        return SingletonHolder.SINGLETON;
+    }
+
+    public void profiling(ClientRequestResult result) {
+        instanceProfiler.profiling(result);
+    }
+
+    public void setInstanceProfiler(ClientInstanceProfiler instanceProfiler) {
+        this.instanceProfiler = instanceProfiler;
+    }
+
+    private static class DefaultClientInstanceProfiler extends AbstractCanalLifeCycle implements ClientInstanceProfiler {
+        @Override
+        public void profiling(ClientRequestResult result) {}
+    }
+
+}

server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfiler.java

@@ -0,0 +1,13 @@
+package com.alibaba.otter.canal.server.netty;
+
+import com.alibaba.otter.canal.common.CanalLifeCycle;
+import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator.ClientRequestResult;
+
+/**
+ * @author Chuanyi Li
+ */
+public interface ClientInstanceProfiler extends CanalLifeCycle {
+
+    void profiling(ClientRequestResult result);
+
+}

server/src/main/java/com/alibaba/otter/canal/server/netty/NettyUtils.java

@@ -74,4 +74,20 @@ public class NettyUtils {
                 .toByteArray(),
             channelFutureListener);
     }
+
+    public static byte[] ackPacket() {
+        return Packet.newBuilder()
+                .setType(CanalPacket.PacketType.ACK)
+                .setBody(Ack.newBuilder().build().toByteString())
+                .build()
+                .toByteArray();
+    }
+
+    public static byte[] errorPacket(int errorCode, String errorMessage) {
+        return Packet.newBuilder()
+                .setType(CanalPacket.PacketType.ACK)
+                .setBody(Ack.newBuilder().setErrorCode(errorCode).setErrorMessage(errorMessage).build().toByteString())
+                .build()
+                .toByteArray();
+    }
 }

server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java

@@ -73,7 +73,7 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
                         MDC.remove("destination");
                     }
                 }
-
+                // 鉴权一次性，暂不统计
                 NettyUtils.ack(ctx.getChannel(), new ChannelFutureListener() {
 
                     public void operationComplete(ChannelFuture future) throws Exception {

server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java

@@ -3,6 +3,7 @@ package com.alibaba.otter.canal.server.netty.handler;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.jboss.netty.buffer.ChannelBuffer;
@@ -51,6 +52,7 @@ public class SessionHandler extends SimpleChannelHandler {
 
     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
         logger.info("message receives in session handler...");
+        long start = System.nanoTime();
         ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
         Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
         ClientIdentity clientIdentity = null;
@@ -74,12 +76,13 @@ public class SessionHandler extends SimpleChannelHandler {
 
                         embeddedServer.subscribe(clientIdentity);
                         // ctx.setAttachment(clientIdentity);// 设置状态数据
-                        NettyUtils.ack(ctx.getChannel(), null);
+                        byte[] ackBytes = NettyUtils.ackPacket();
+                        NettyUtils.write(ctx.getChannel(), ackBytes, new ChannelFutureAggregator(sub.getDestination(),
+                                sub, packet.getType(), ackBytes.length, System.nanoTime() - start));
                     } else {
-                        NettyUtils.error(401,
-                            MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage(),
-                            ctx.getChannel(),
-                            null);
+                        byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage());
+                        NettyUtils.write(ctx.getChannel(), errorBytes ,new ChannelFutureAggregator(sub.getDestination(),
+                                sub, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 401));
                     }
                     break;
                 case UNSUBSCRIPTION:
@@ -91,12 +94,13 @@ public class SessionHandler extends SimpleChannelHandler {
                         MDC.put("destination", clientIdentity.getDestination());
                         embeddedServer.unsubscribe(clientIdentity);
                         stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭
-                        NettyUtils.ack(ctx.getChannel(), null);
+                        byte[] ackBytes = NettyUtils.ackPacket();
+                        NettyUtils.write(ctx.getChannel(), ackBytes, new ChannelFutureAggregator(unsub.getDestination(),
+                                unsub, packet.getType(), ackBytes.length, System.nanoTime() - start));
                     } else {
-                        NettyUtils.error(401,
-                            MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage(),
-                            ctx.getChannel(),
-                            null);
+                        byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage());
+                        NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(unsub.getDestination(),
+                                unsub, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 401));
                     }
                     break;
                 case GET:
@@ -171,7 +175,8 @@ public class SessionHandler extends SimpleChannelHandler {
                                 output.writeBytes(2, rowEntries.get(i));
                             }
                             output.checkNoSpaceLeft();
-                            NettyUtils.write(ctx.getChannel(), body, null);
+                            NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
+                                    get, packet.getType(), body.length, System.nanoTime() - start, message.getId() == -1));
                             
                             // output.flush();
                             // byteBuffer.flip();
@@ -192,14 +197,14 @@ public class SessionHandler extends SimpleChannelHandler {
                                     }
                                 }
                             }
-                            packetBuilder.setBody(messageBuilder.build().toByteString());
-                            NettyUtils.write(ctx.getChannel(), packetBuilder.build().toByteArray(), null);// 输出数据
+                            byte[] body = packetBuilder.setBody(messageBuilder.build().toByteString()).build().toByteArray();
+                            NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
+                                    get, packet.getType(), body.length, System.nanoTime() - start, message.getId() == -1));// 输出数据
                         }
                     } else {
-                        NettyUtils.error(401,
-                            MessageFormatter.format("destination or clientId is null", get.toString()).getMessage(),
-                            ctx.getChannel(),
-                            null);
+                        byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", get.toString()).getMessage());
+                        NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(get.getDestination(),
+                                get, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 401));
                     }
                     break;
                 case CLIENTACK:
@@ -207,10 +212,9 @@ public class SessionHandler extends SimpleChannelHandler {
                     MDC.put("destination", ack.getDestination());
                     if (StringUtils.isNotEmpty(ack.getDestination()) && StringUtils.isNotEmpty(ack.getClientId())) {
                         if (ack.getBatchId() == 0L) {
-                            NettyUtils.error(402,
-                                MessageFormatter.format("batchId should assign value", ack.toString()).getMessage(),
-                                ctx.getChannel(),
-                                null);
+                            byte[] errorBytes = NettyUtils.errorPacket(402, MessageFormatter.format("batchId should assign value", ack.toString()).getMessage());
+                            NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ack.getDestination(),
+                                    ack, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 402));
                         } else if (ack.getBatchId() == -1L) { // -1代表上一次get没有数据，直接忽略之
                             // donothing
                         } else {
@@ -218,10 +222,9 @@ public class SessionHandler extends SimpleChannelHandler {
                             embeddedServer.ack(clientIdentity, ack.getBatchId());
                         }
                     } else {
-                        NettyUtils.error(401,
-                            MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage(),
-                            ctx.getChannel(),
-                            null);
+                        byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage());
+                        NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ack.getDestination(),
+                                ack, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 401));
                     }
                     break;
                 case CLIENTROLLBACK:
@@ -237,25 +240,23 @@ public class SessionHandler extends SimpleChannelHandler {
                             embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
                         }
                     } else {
-                        NettyUtils.error(401,
-                            MessageFormatter.format("destination or clientId is null", rollback.toString())
-                                .getMessage(),
-                            ctx.getChannel(),
-                            null);
+                        byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", rollback.toString()).getMessage());
+                        NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(rollback.getDestination(),
+                                rollback, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 401));
                     }
                     break;
                 default:
-                    NettyUtils.error(400, MessageFormatter.format("packet type={} is NOT supported!", packet.getType())
-                        .getMessage(), ctx.getChannel(), null);
+                    byte[] errorBytes = NettyUtils.errorPacket(400, MessageFormatter.format("packet type={} is NOT supported!", packet.getType()).getMessage());
+                    NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel().getRemoteAddress().toString(),
+                            null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
                     break;
             }
         } catch (Throwable exception) {
-            NettyUtils.error(400,
-                MessageFormatter.format("something goes wrong with channel:{}, exception={}",
+            byte[] errorBytes = NettyUtils.errorPacket(400, MessageFormatter.format("something goes wrong with channel:{}, exception={}",
                     ctx.getChannel(),
-                    ExceptionUtils.getStackTrace(exception)).getMessage(),
-                ctx.getChannel(),
-                null);
+                    ExceptionUtils.getStackTrace(exception)).getMessage());
+            NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel().getRemoteAddress().toString(),
+                    null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
         } finally {
             MDC.remove("destination");
         }

server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java

@@ -0,0 +1,179 @@
+package com.alibaba.otter.canal.server.netty.listener;
+
+import com.alibaba.otter.canal.protocol.CanalPacket;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.GeneratedMessage;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+
+import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler.profiler;
+
+/**
+ * @author Chuanyi Li
+ */
+public class ChannelFutureAggregator implements ChannelFutureListener {
+
+    private ClientRequestResult result;
+
+    public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, boolean empty) {
+        this(destination, request, type, amount, latency, empty, (short) 0);
+    }
+
+    public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency) {
+        this(destination, request, type, amount, latency, false, (short) 0);
+    }
+
+    public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, short errorCode) {
+        this(destination, request, type, amount, latency, false, errorCode);
+    }
+
+    private ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, boolean empty, short errorCode) {
+        this.result = new ClientRequestResult.Builder()
+                .destination(destination)
+                .type(type)
+                .request(request)
+                .amount(amount)
+                .latency(latency)
+                .errorCode(errorCode)
+                .empty(empty)
+                .build();
+    }
+
+    @Override
+    public void operationComplete(ChannelFuture future) {
+        // profiling after I/O operation
+        if (future.getCause() != null) {
+            result.channelError = future.getCause();
+        }
+        profiler().profiling(result);
+    }
+
+    /**
+     * Client request result pojo
+     */
+    public static class ClientRequestResult {
+
+        private String                 destination;
+        private CanalPacket.PacketType type;
+        private GeneratedMessage       request;
+        private int                    amount;
+        private long                   latency;
+        private short                  errorCode;
+        private boolean                empty;
+        private Throwable              channelError;
+
+        private ClientRequestResult() {}
+
+        private ClientRequestResult(Builder builder) {
+            this.destination = Preconditions.checkNotNull(builder.destination);
+            this.type = Preconditions.checkNotNull(builder.type);
+            this.request = builder.request;
+            this.amount = builder.amount;
+            this.latency = builder.latency;
+            this.errorCode = builder.errorCode;
+            this.empty = builder.empty;
+            this.channelError = builder.channelError;
+        }
+
+        // auto-generated
+        public static class Builder {
+
+            private String                 destination;
+            private CanalPacket.PacketType type;
+            private GeneratedMessage       request;
+            private int                    amount;
+            private long                   latency;
+            private short                  errorCode;
+            private boolean                empty;
+            private Throwable              channelError;
+
+            Builder destination(String destination) {
+                this.destination = destination;
+                return this;
+            }
+
+            Builder type(CanalPacket.PacketType type) {
+                this.type = type;
+                return this;
+            }
+
+            Builder request(GeneratedMessage request) {
+                this.request = request;
+                return this;
+            }
+
+            Builder amount(int amount) {
+                this.amount = amount;
+                return this;
+            }
+
+            Builder latency(long latency) {
+                this.latency = latency;
+                return this;
+            }
+
+            Builder errorCode(short errorCode) {
+                this.errorCode = errorCode;
+                return this;
+            }
+
+            Builder empty(boolean empty) {
+                this.empty = empty;
+                return this;
+            }
+
+            public Builder channelError(Throwable channelError) {
+                this.channelError = channelError;
+                return this;
+            }
+
+            public Builder fromPrototype(ClientRequestResult prototype) {
+                destination = prototype.destination;
+                type = prototype.type;
+                request = prototype.request;
+                amount = prototype.amount;
+                latency = prototype.latency;
+                errorCode = prototype.errorCode;
+                empty = prototype.empty;
+                channelError = prototype.channelError;
+                return this;
+            }
+
+            ClientRequestResult build() {
+                return new ClientRequestResult(this);
+            }
+        }
+        // getters
+        public String getDestination() {
+            return destination;
+        }
+
+        public CanalPacket.PacketType getType() {
+            return type;
+        }
+
+        public GeneratedMessage getRequest() {
+            return request;
+        }
+
+        public int getAmount() {
+            return amount;
+        }
+
+        public long getLatency() {
+            return latency;
+        }
+
+        public short getErrorCode() {
+            return errorCode;
+        }
+
+        public boolean getEmpty() {
+            return empty;
+        }
+
+        public Throwable getChannelError() {
+            return channelError;
+        }
+    }
+}

sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java

@@ -39,6 +39,7 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
     protected long                 emptyTransctionThresold       = 8192;                                         // 超过1024个事务头，输出一个
     protected volatile long        lastEmptyTransactionTimestamp = 0L;
     protected AtomicLong           lastEmptyTransactionCount     = new AtomicLong(0L);
+    private AtomicLong             eventsSinkBlockingTime        = new AtomicLong(0L);
 
     public EntryEventSink(){
         addHandler(new HeartBeatEntryEventHandler());
@@ -147,16 +148,27 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
         for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
             events = handler.before(events);
         }
-
+        long blockingStart = 0L;
         int fullTimes = 0;
         do {
             if (eventStore.tryPut(events)) {
+                if (fullTimes > 0) {
+                    eventsSinkBlockingTime.addAndGet(System.nanoTime() - blockingStart);
+                }
                 for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
                     events = handler.after(events);
                 }
                 return true;
             } else {
+                if (fullTimes == 0) {
+                    blockingStart = System.nanoTime();
+                }
                 applyWait(++fullTimes);
+                if (fullTimes % 100 == 0) {
+                    long nextStart = System.nanoTime();
+                    eventsSinkBlockingTime.addAndGet(nextStart - blockingStart);
+                    blockingStart = nextStart;
+                }
             }
 
             for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
@@ -202,4 +214,8 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
         this.emptyTransctionThresold = emptyTransctionThresold;
     }
 
+    public AtomicLong getEventsSinkBlockingTime() {
+        return eventsSinkBlockingTime;
+    }
+
 }

store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java

@@ -562,4 +562,23 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
         this.ddlIsolation = ddlIsolation;
     }
 
+    public AtomicLong getPutSequence() {
+        return putSequence;
+    }
+
+    public AtomicLong getAckSequence() {
+        return ackSequence;
+    }
+
+    public AtomicLong getPutMemSize() {
+        return putMemSize;
+    }
+
+    public AtomicLong getAckMemSize() {
+        return ackMemSize;
+    }
+
+    public BatchMode getBatchMode() {
+        return batchMode;
+    }
 }

store/src/main/java/com/alibaba/otter/canal/store/model/Event.java

@@ -1,6 +1,7 @@
 package com.alibaba.otter.canal.store.model;
 
 import java.io.Serializable;
+import java.util.List;
 
 import org.apache.commons.lang.builder.ToStringBuilder;
 
@@ -31,6 +32,7 @@ public class Event implements Serializable {
     private EventType         eventType;
     private String            gtid;
     private long              rawLength;
+    private int               rowsCount;
 
     public Event(){
     }
@@ -47,6 +49,17 @@ public class Event implements Serializable {
         // build raw
         this.rawEntry = entry.toByteString();
         this.rawLength = rawEntry.size();
+        if (entryType == EntryType.ROWDATA) {
+            List<CanalEntry.Pair> props = entry.getHeader().getPropsList();
+            if (props != null) {
+                for (CanalEntry.Pair p : props) {
+                    if ("rowsCount".equals(p.getKey())) {
+                        rowsCount = Integer.parseInt(p.getValue());
+                        break;
+                    }
+                }
+            }
+        }
     }
 
     public LogIdentity getLogIdentity() {
@@ -129,7 +142,16 @@ public class Event implements Serializable {
         this.eventType = eventType;
     }
 
+    public int getRowsCount() {
+        return rowsCount;
+    }
+
+    public void setRowsCount(int rowsCount) {
+        this.rowsCount = rowsCount;
+    }
+
     public String toString() {
         return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE);
     }
+
 }