chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Enhance heartbeat, rowscount.

Browse Source
This commit is contained in:
Chuanyi Li
2018-08-08 22:53:08 +08:00
parent bf9dc0073d
commit 760f2dda80
8 changed files with 12743 additions and 12927 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
prometheus/pom.xml
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>

31
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java
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;
}

13
prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java
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;