wyl/canal
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Enhance heartbeat, rowscount.

Browse Source
This commit is contained in:
Chuanyi Li
2018-08-09 23:54:53 +08:00
parent d7bbb0d1a7
commit ec00792475
8 changed files with 12743 additions and 12927 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBuffer.java
+5
View File
@@ -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/dbsync/LogEventConvert.java
+12 -15
View File
@@ -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();
prometheus/pom.xml
-11
View File
@@ -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/impl/EntryCollector.java
+25 -6
View File
@@ -7,6 +7,7 @@ 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;
@@ -26,11 +27,15 @@ import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABEL
*/
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 DELAY_HELP = "Traffic delay of canal instance in milliseconds";
private static final String TRANSACTION_HELP = "Transactions counter of canal instance";
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() {}
@@ -50,16 +55,24 @@ public class EntryCollector extends Collector implements InstanceRegistry {
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) {
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;
}
@@ -76,8 +89,12 @@ public class EntryCollector extends Collector implements InstanceRegistry {
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);
@@ -128,6 +145,8 @@ public class EntryCollector extends Collector implements InstanceRegistry {
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/PrometheusCanalEventDownStreamHandler.java
+5 -8
View File
@@ -34,7 +34,8 @@ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDow
case ROWDATA: {
long exec = e.getExecuteTime();
if (exec > 0) localExecTime = exec;
// TODO 当前proto无法直接获得荣威change的变更行数（需要parse），可考虑放到header里面
rowEventCounter.incrementAndGet();
rowsCounter.addAndGet(e.getRowsCount());
break;
}
case TRANSACTIONEND: {
@@ -44,14 +45,10 @@ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDow
break;
}
case HEARTBEAT:
// 发现canal自己的heartbeat是带有execTime的
// TODO 确认一下不是canal自己产生的
CanalEntry.EventType eventType = e.getEventType();
// 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.
if (eventType == CanalEntry.EventType.MHEARTBEAT) {
localExecTime = System.currentTimeMillis();
}
break;
default:
break;
protocol/src/main/java/com/alibaba/otter/canal/protocol/CanalEntry.java
+12671 -12886
View File
File diff suppressed because it is too large Load Diff
protocol/src/main/java/com/alibaba/otter/canal/protocol/EntryProtocol.proto
+3 -1
View File
@@ -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;
}
}
store/src/main/java/com/alibaba/otter/canal/store/model/Event.java
+22
View File
@@ -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);
}
}