diff --git a/common/src/main/java/com/alibaba/otter/canal/common/utils/SerializedLongAdder.java b/common/src/main/java/com/alibaba/otter/canal/common/utils/SerializedLongAdder.java new file mode 100644 index 00000000..4d86a99e --- /dev/null +++ b/common/src/main/java/com/alibaba/otter/canal/common/utils/SerializedLongAdder.java @@ -0,0 +1,31 @@ +package com.alibaba.otter.canal.common.utils; + +/** + * 用于dump记录counter + * 这种场景下， + * 写操作>>读操作 + * dump线程为串行，读取counter时，并不严格要求最新值可见，使用普通long即可。 + * + * @author Chuanyi Li + */ +public class SerializedLongAdder { + + private long value = 0; + + public SerializedLongAdder(long initial) { + this.value = initial; + } + + public void add(long x) { + value += x; + } + + public void increment() { + value ++; + } + + public long get() { + return value; + } + +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 0034ba2b..62096b6a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -9,6 +9,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; +import com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.RandomUtils; @@ -100,6 +101,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle + protected abstract BinlogParser buildParser(); protected abstract ErosaConnection buildErosaConnection(); @@ -643,4 +645,5 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle public void setServerId(long serverId) { this.serverId = serverId; } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 17eb08f4..5209b178 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.nio.charset.Charset; +import com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,6 +39,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected boolean filterRows = false; protected boolean filterTableError = false; protected boolean useDruidDdlFilter = true; + private final SerializedLongAdder eventsPublishBlockingTime = new SerializedLongAdder(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 ========================= @@ -203,5 +207,9 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { public void setTableMetaTSDBFactory(TableMetaTSDBFactory tableMetaTSDBFactory) { this.tableMetaTSDBFactory = tableMetaTSDBFactory; } + + public SerializedLongAdder getEventsPublishBlockingTime() { + return this.eventsPublishBlockingTime; + } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 5f681a54..c0c658fb 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -8,6 +8,7 @@ import java.nio.charset.Charset; import java.util.List; import java.util.concurrent.TimeUnit; +import com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.math.NumberUtils; import org.slf4j.Logger; @@ -39,18 +40,19 @@ 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小时 + private SerializedLongAdder receivedBinlogBytes; public MysqlConnection(){ } @@ -126,6 +128,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); @@ -148,6 +151,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); @@ -176,6 +180,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); @@ -206,6 +211,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)) { @@ -232,6 +238,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)) { @@ -346,7 +353,6 @@ public class MysqlConnection implements ErosaConnection { *
  • net_read_timeout
    • * * - * @param channel * @throws IOException */ private void updateSettings() throws IOException { @@ -465,6 +471,14 @@ public class MysqlConnection implements ErosaConnection { } } + private void accumulateReceivedBytes(long x) { + if (receivedBinlogBytes != null) { + receivedBinlogBytes.add(x); + } + } + + + public static enum BinlogFormat { STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED"); @@ -604,4 +618,8 @@ public class MysqlConnection implements ErosaConnection { this.authInfo = authInfo; } + public void setReceivedBinlogBytes(SerializedLongAdder receivedBinlogBytes) { + this.receivedBinlogBytes = receivedBinlogBytes; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 8faaaa61..2b1e65e2 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -10,6 +10,7 @@ import java.util.Map; import java.util.TimerTask; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.apache.commons.lang.StringUtils; import org.springframework.util.CollectionUtils; @@ -68,8 +69,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 SerializedLongAdder receivedBinlogBytes = new SerializedLongAdder(0L); protected ErosaConnection buildErosaConnection() { return buildMysqlConnection(this.runningInfo); @@ -313,6 +317,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(); @@ -910,4 +915,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE this.dumpErrorCountThreshold = dumpErrorCountThreshold; } + + + public SerializedLongAdder getReceivedBinlogBytes() { + return this.receivedBinlogBytes; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 6acc5056..e348c386 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -4,6 +4,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.locks.LockSupport; +import com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.apache.commons.lang.StringUtils; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; @@ -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 SerializedLongAdder 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,34 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement event.setBinlogFileName(binlogFileName); } disruptorMsgBuffer.publish(next); + if (fullTimes > 0) { + eventsPublishBlockingTime.add(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(); } } 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 +212,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement start(); } + private class SimpleParserStage implements EventHandler, LifecycleAware { private LogDecoder decoder; @@ -433,4 +457,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement this.connection = connection; } + public void setEventsPublishBlockingTime(SerializedLongAdder eventsPublishBlockingTime) { + this.eventsPublishBlockingTime = eventsPublishBlockingTime; + } + } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalInstanceExports.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalInstanceExports.java index 8acf40b8..19759154 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalInstanceExports.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalInstanceExports.java @@ -1,9 +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.prometheus.impl.*; import com.alibaba.otter.canal.sink.CanalEventSink; import com.alibaba.otter.canal.sink.entry.EntryEventSink; import com.alibaba.otter.canal.store.CanalStoreException; @@ -15,30 +13,31 @@ import org.slf4j.LoggerFactory; import java.util.Collections; import java.util.List; +import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler.profiler; + /** * @author Chuanyi Li */ public class CanalInstanceExports { - private static final Logger logger = LoggerFactory.getLogger(CanalInstanceExports.class); - - public static final String[] labels = {"destination"}; - - public static final List labelList = Collections.singletonList(labels[0]); - + private static final Logger logger = LoggerFactory.getLogger(CanalInstanceExports.class); + public static final String DESTINATION = "destination"; + public static final String[] DEST_LABELS = {DESTINATION}; + public static final List DEST_LABEL_LIST = Collections.singletonList(DESTINATION); private final String destination; - private Collector storeCollector; - private Collector delayCollector; - private Collector metaCollector; + private Collector sinkCollector; + private Collector parserCollector; private CanalInstanceExports(CanalInstance instance) { this.destination = instance.getDestination(); - initDelayGauge(instance); + initEventsMetrics(instance); initStoreCollector(instance); initMetaCollector(instance); + initSinkCollector(instance); + initParserCollector(instance); } @@ -48,6 +47,7 @@ public class CanalInstanceExports { } void register() { + profiler().start(destination); if (delayCollector != null) { delayCollector.register(); } @@ -57,9 +57,16 @@ public class CanalInstanceExports { if (metaCollector != null) { metaCollector.register(); } + if (sinkCollector != null) { + sinkCollector.register(); + } + if (parserCollector != null) { + parserCollector.register(); + } } void unregister() { + profiler().stop(destination); if (delayCollector != null) { CollectorRegistry.defaultRegistry.unregister(delayCollector); } @@ -69,9 +76,15 @@ public class CanalInstanceExports { if (metaCollector != null) { CollectorRegistry.defaultRegistry.unregister(metaCollector); } + if (sinkCollector != null) { + CollectorRegistry.defaultRegistry.unregister(sinkCollector); + } + if (parserCollector != null) { + CollectorRegistry.defaultRegistry.unregister(parserCollector); + } } - private void initDelayGauge(CanalInstance instance) { + private void initEventsMetrics(CanalInstance instance) { CanalEventSink sink = instance.getEventSink(); if (sink instanceof EntryEventSink) { EntryEventSink entryEventSink = (EntryEventSink) sink; @@ -95,4 +108,12 @@ public class CanalInstanceExports { private void initMetaCollector(CanalInstance instance) { metaCollector = new InstanceMetaCollector(instance); } + + private void initSinkCollector(CanalInstance instance) { + sinkCollector = new EntrySinkCollector(instance.getEventSink(), instance.getDestination()); + } + + private void initParserCollector(CanalInstance instance) { + parserCollector = new MysqlParserCollector(instance.getEventParser(), instance.getDestination()); + } } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalServerExports.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalServerExports.java index 8eef79ba..ded44fb8 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalServerExports.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalServerExports.java @@ -1,7 +1,5 @@ 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 @@ -12,8 +10,6 @@ public class CanalServerExports { public static synchronized void initialize() { if (!initialized) { - InboundThroughputAspect.getCollector().register(); - OutboundThroughputAspect.getCollector().register(); initialized = true; } } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java index c7942235..2aa7dcd0 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/PrometheusService.java @@ -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.PrometheusClientInstanceProfilerFactory; +import com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler; import com.alibaba.otter.canal.spi.CanalMetricsService; import io.prometheus.client.exporter.HTTPServer; import io.prometheus.client.hotspot.DefaultExports; @@ -11,6 +13,9 @@ import java.io.IOException; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler.DISABLED; +import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler.profiler; + /** * @author Chuanyi Li */ @@ -50,6 +55,7 @@ public class PrometheusService implements CanalMetricsService { DefaultExports.initialize(); // Canal server level exports CanalServerExports.initialize(); + profiler().setInstanceProfilerFactory(new PrometheusClientInstanceProfilerFactory()); } catch (Throwable t) { logger.warn("Unable to initialize server exports.", t); } @@ -66,6 +72,7 @@ public class PrometheusService implements CanalMetricsService { for (CanalInstanceExports ie : exports.values()) { ie.unregister(); } + profiler().setInstanceProfilerFactory(DISABLED); if (server != null) { server.stop(); } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntrySinkCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntrySinkCollector.java new file mode 100644 index 00000000..54775842 --- /dev/null +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntrySinkCollector.java @@ -0,0 +1,45 @@ +package com.alibaba.otter.canal.prometheus.impl; + +import com.alibaba.otter.canal.sink.CanalEventSink; +import com.alibaba.otter.canal.sink.entry.EntryEventSink; +import io.prometheus.client.Collector; +import io.prometheus.client.CounterMetricFamily; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABEL_LIST; + +/** + * @author Chuanyi Li + */ +public class EntrySinkCollector extends Collector { + + private static final String SINK_BLOCKING_TIME = "canal_instance_sink_blocking_time"; + private final String sinkBlockTimeHelp; + private final List destLabelValues; + private final AtomicLong eventsSinkBlockingTime; + + + public EntrySinkCollector(CanalEventSink sink, String destination) { + if (!(sink instanceof EntryEventSink)) { + throw new IllegalArgumentException("CanalEventSink must be EntryEventSink"); + } + this.destLabelValues = Collections.singletonList(destination); + EntryEventSink entrySink = (EntryEventSink) sink; + this.sinkBlockTimeHelp = "Sink blocking time of instance " + destination; + eventsSinkBlockingTime = entrySink.getEventsSinkBlockingTime(); + } + + @Override public List collect() { + List mfs = new ArrayList(); + CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME, + sinkBlockTimeHelp, DEST_LABEL_LIST); + blockingCounter.addMetric(destLabelValues, eventsSinkBlockingTime.doubleValue()); + mfs.add(blockingCounter); + return mfs; + } + +} diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InboundThroughputAspect.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InboundThroughputAspect.java deleted file mode 100644 index eabc256c..00000000 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InboundThroughputAspect.java +++ /dev/null @@ -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 collect() { - List mfs = new ArrayList(); - CounterMetricFamily bytes = new CounterMetricFamily("canal_net_inbound_bytes", - "Total socket inbound bytes of canal server.", - total.get()); - mfs.add(bytes); - return mfs; - } - } - -} diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InstanceMetaCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InstanceMetaCollector.java index 5562b8fd..2e8b7fc4 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InstanceMetaCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/InstanceMetaCollector.java @@ -55,7 +55,7 @@ public class InstanceMetaCollector extends Collector { List subs = metaManager.listAllSubscribeInfo(destination); GaugeMetricFamily subscriptions = new GaugeMetricFamily( "canal_instance_subscription", - subsHelp, CanalInstanceExports.labelList); + subsHelp, CanalInstanceExports.DEST_LABEL_LIST); subscriptions.addMetric(Arrays.asList(destination), subs.size()); mfs.add(subscriptions); } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MemoryStoreCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MemoryStoreCollector.java index 79fdca4a..4c212c4c 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MemoryStoreCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MemoryStoreCollector.java @@ -20,15 +20,10 @@ import java.util.concurrent.atomic.AtomicLong; public class MemoryStoreCollector extends Collector { private static final Class 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) { @@ -47,11 +42,11 @@ public class MemoryStoreCollector extends Collector { public List collect() { List mfs = new ArrayList(); CounterMetricFamily put = new CounterMetricFamily("canal_instance_store_produce_seq", - putHelp, Arrays.asList(CanalInstanceExports.labels)); + putHelp, Arrays.asList(CanalInstanceExports.DEST_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)); + ackHelp, Arrays.asList(CanalInstanceExports.DEST_LABELS)); ack.addMetric(Collections.singletonList(destination), ackSequence.doubleValue()); mfs.add(ack); return mfs; diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MysqlParserCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MysqlParserCollector.java new file mode 100644 index 00000000..02d35209 --- /dev/null +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MysqlParserCollector.java @@ -0,0 +1,74 @@ +package com.alibaba.otter.canal.prometheus.impl; + +import com.alibaba.otter.canal.common.utils.SerializedLongAdder; +import com.alibaba.otter.canal.parse.CanalEventParser; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; +import io.prometheus.client.Collector; +import io.prometheus.client.CounterMetricFamily; +import io.prometheus.client.GaugeMetricFamily; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DESTINATION; +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABEL_LIST; + +/** + * @author Chuanyi Li + */ +public class MysqlParserCollector extends Collector { + + private static final String PUBLISH_BLOCKING_TIME = "canal_instance_publish_blocking_time"; + private static final String RECEIVED_BINLOG_BYTES = "canal_instance_received_binlog_bytes"; + private static final String PARSER_MODE = "canal_instance_parser_mode"; + private static final String MODE_LABEL = "parallel"; + private final List destLabelValues; + private final List modeLabelValues; + private final SerializedLongAdder receivedBinlogBytes; + private final SerializedLongAdder eventsPublishBlockingTime; + private final String publishBlockingTimeHelp; + private final String receivedBinlogBytesHelp; + private final String modeHelp; + private final List modeLabels; + private final Boolean isParallel; + + public MysqlParserCollector(CanalEventParser parser, String destination) { + if (!(parser instanceof MysqlEventParser)) { + throw new IllegalArgumentException("CanalEventParser must be MysqlEventParser"); + } + this.destLabelValues = Collections.singletonList(destination); + this.eventsPublishBlockingTime = ((MysqlEventParser)parser).getEventsPublishBlockingTime(); + this.publishBlockingTimeHelp = "Blocking time of publishing of instance " + destination; + this.receivedBinlogBytes = ((MysqlEventParser)parser).getReceivedBinlogBytes(); + this.receivedBinlogBytesHelp = "Received binlog bytes of instance" + destination; + this.isParallel = ((MysqlEventParser)parser).isParallel(); + this.modeHelp = "Parser mode of instance" + destination; + this.modeLabels = Arrays.asList(DESTINATION, MODE_LABEL); + this.modeLabelValues = Arrays.asList(destination, isParallel.toString()); + } + + @Override public List collect() { + List mfs = new ArrayList(); + // canal_instance_received_binlog_bytes + CounterMetricFamily bytesCounter = new CounterMetricFamily(RECEIVED_BINLOG_BYTES, + receivedBinlogBytesHelp, DEST_LABEL_LIST); + bytesCounter.addMetric(destLabelValues, receivedBinlogBytes.get()); + mfs.add(bytesCounter); + // canal_instance_parser_mode + GaugeMetricFamily modeGauge = new GaugeMetricFamily(PARSER_MODE, + modeHelp, + modeLabels); + modeGauge.addMetric(modeLabelValues, 1); + mfs.add(modeGauge); + // canal_instance_publish_blocking_time + if (isParallel) { + CounterMetricFamily blockingCounter = new CounterMetricFamily(PUBLISH_BLOCKING_TIME, + publishBlockingTimeHelp, DEST_LABEL_LIST); + blockingCounter.addMetric(destLabelValues, eventsPublishBlockingTime.get()); + mfs.add(blockingCounter); + } + return mfs; + } +} diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/OutboundThroughputAspect.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/OutboundThroughputAspect.java deleted file mode 100644 index 60dd808d..00000000 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/OutboundThroughputAspect.java +++ /dev/null @@ -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 collect() { - List mfs = new ArrayList(); - CounterMetricFamily bytes = new CounterMetricFamily("canal_net_outbound_bytes", - "Total socket outbound bytes of canal server.", - total.get()); - mfs.add(bytes); - return mfs; - } - } -} diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java index c55a7eac..47d4824d 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusCanalEventDownStreamHandler.java @@ -36,7 +36,7 @@ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDow GaugeMetricFamily delay = new GaugeMetricFamily( DELAY_NAME, delayHelpName, - CanalInstanceExports.labelList); + CanalInstanceExports.DEST_LABEL_LIST); double d = 0.0; if (latestExecuteTime > 0) { d = now - latestExecuteTime; diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfiler.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfiler.java new file mode 100644 index 00000000..d3413455 --- /dev/null +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfiler.java @@ -0,0 +1,134 @@ +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.DESTINATION; +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS; + +/** + * @author Chuanyi Li + */ +public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler { + + 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 volatile boolean running = false; + private final String destination; + private final String[] destLabelValues; + private final Counter outboundCounter; + private final Counter packetsCounter; + private final Counter emptyBatchesCounter; + private final Counter errorsCounter; + private final Histogram responseLatency; + + public PrometheusClientInstanceProfiler(String destination) { + this.destination = destination; + this.destLabelValues = new String[]{destination}; + this.outboundCounter = Counter.build() + .labelNames(DEST_LABELS) + .name(OUTBOUND_BYTES) + .help("Send bytes to client of instance " + destination) + .create(); + this.packetsCounter = Counter.build() + .labelNames(new String[]{DESTINATION, "packetType"}) + .name(PACKET_TYPE) + .help("Send packets to client of instance " + destination) + .create(); + this.emptyBatchesCounter = Counter.build() + .labelNames(DEST_LABELS) + .name(EMPTY_BATCHES) + .help("Send empty batches to client of instance " + destination) + .create(); + this.errorsCounter = Counter.build() + .labelNames(new String[]{DESTINATION, "errorCode"}) + .name(ERRORS) + .help("Client request errors of instance " + destination) + .create(); + this.responseLatency = Histogram.build() + .labelNames(DEST_LABELS) + .name(LATENCY) + .help("Client request latency of instance " + destination) + // buckets in milliseconds + .buckets(1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0) + .create(); + } + + @Override + public String getDestination() { + return this.destination; + } + + @Override + public void profiling(ClientRequestResult result) { + PacketType type = result.getType(); + outboundCounter.labels(destLabelValues).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(destLabelValues).observe(latency / 1000000); + switch (type) { + case GET: + boolean empty = result.getEmpty(); + if (empty) { + emptyBatchesCounter.labels(destLabelValues).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; + } +} diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfilerFactory.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfilerFactory.java new file mode 100644 index 00000000..6f80c221 --- /dev/null +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/PrometheusClientInstanceProfilerFactory.java @@ -0,0 +1,15 @@ +package com.alibaba.otter.canal.prometheus.impl; + +import com.alibaba.otter.canal.server.netty.ClientInstanceProfiler; +import com.alibaba.otter.canal.server.netty.ClientInstanceProfilerFactory; + +/** + * @author Chuanyi Li + */ +public class PrometheusClientInstanceProfilerFactory implements ClientInstanceProfilerFactory { + + @Override + public ClientInstanceProfiler create(String destination) { + return new PrometheusClientInstanceProfiler(destination); + } +} diff --git a/prometheus/src/main/resources/META-INF/aop.xml b/prometheus/src/main/resources/META-INF/aop.xml deleted file mode 100644 index 8bd31d0c..00000000 --- a/prometheus/src/main/resources/META-INF/aop.xml +++ /dev/null @@ -1,11 +0,0 @@ - - - - - - - - - - - diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNettyProfiler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNettyProfiler.java index 9e9d9a39..cfd05ee6 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNettyProfiler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNettyProfiler.java @@ -10,7 +10,7 @@ import java.util.concurrent.ConcurrentMap; */ public class CanalServerWithNettyProfiler { - private static final ClientInstanceProfilerFactory DISABLED = new DefaultClientInstanceProfilerFactory(); + public static final ClientInstanceProfilerFactory DISABLED = new DefaultClientInstanceProfilerFactory(); private volatile ClientInstanceProfilerFactory factory; private final ConcurrentMap cliPfs; private final CanalServerWithEmbedded server; @@ -29,34 +29,47 @@ public class CanalServerWithNettyProfiler { return SingletonHolder.SINGLETON; } - public void profiling(String dest, ClientRequestResult result) { + public void profiling(String destination, ClientRequestResult result) { if (isDisabled()) { return; } - ClientInstanceProfiler profiler = tryGet(dest); + ClientInstanceProfiler profiler = cliPfs.get(destination); if (profiler != null) { profiler.profiling(result); } } - /** - * Remove instance profiler for specified instance. - * Only accepted while instance is not running. - * @param dest canal instance destination - */ - public void remove(String dest) throws IllegalStateException { + public void start(String destination) { if (isDisabled()) { return; } - synchronized (cliPfs) { - if (server.isStart(dest)) { - throw new IllegalStateException("Instance profiler should not be removed while running."); - } - cliPfs.remove(dest); + if (server.isStart(destination)) { + throw new IllegalStateException("Instance profiler should not be start while running."); + } + ClientInstanceProfiler profiler = factory.create(destination); + profiler.start(); + cliPfs.put(destination, profiler); + } + + /** + * Remove instance profiler for specified instance. + * Only accepted while instance is not running. + * @param destination canal instance destination + */ + public void stop(String destination) { + if (isDisabled()) { + return; + } + if (server.isStart(destination)) { + throw new IllegalStateException("Instance profiler should not be stop while running."); + } + ClientInstanceProfiler profiler = cliPfs.remove(destination); + if (profiler != null && profiler.isStart()) { + profiler.stop(); } } - public void setFactory(ClientInstanceProfilerFactory factory) { + public void setInstanceProfilerFactory(ClientInstanceProfilerFactory factory) { this.factory = factory; } @@ -64,15 +77,18 @@ public class CanalServerWithNettyProfiler { return factory == DISABLED || factory == null; } - private ClientInstanceProfiler tryGet(String dest) { + private ClientInstanceProfiler tryGet(String destination) { //try fast get - ClientInstanceProfiler profiler = cliPfs.get(dest); + ClientInstanceProfiler profiler = cliPfs.get(destination); if (profiler == null) { synchronized (cliPfs) { - if (server.isStart(dest)) { + if (server.isStart(destination)) { // avoid overwriting - cliPfs.putIfAbsent(dest, factory.create()); - profiler = cliPfs.get(dest); + cliPfs.putIfAbsent(destination, factory.create(destination)); + profiler = cliPfs.get(destination); + if (!profiler.isStart()) { + profiler.start(); + } } } } @@ -81,7 +97,7 @@ public class CanalServerWithNettyProfiler { private static class DefaultClientInstanceProfilerFactory implements ClientInstanceProfilerFactory { @Override - public ClientInstanceProfiler create() { + public ClientInstanceProfiler create(String destination) { throw new UnsupportedOperationException(); } } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfiler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfiler.java index 6c9ffc8b..bd4e2dbc 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfiler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfiler.java @@ -1,11 +1,12 @@ package com.alibaba.otter.canal.server.netty; +import com.alibaba.otter.canal.common.CanalLifeCycle; import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator; /** * @author Chuanyi Li */ -public interface ClientInstanceProfiler { +public interface ClientInstanceProfiler extends CanalLifeCycle { String getDestination(); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfilerFactory.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfilerFactory.java index 30075ba1..12b54bc6 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfilerFactory.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/ClientInstanceProfilerFactory.java @@ -5,6 +5,6 @@ package com.alibaba.otter.canal.server.netty; */ public interface ClientInstanceProfilerFactory { - ClientInstanceProfiler create(); + ClientInstanceProfiler create(String destination); } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index 7ea861d4..2d09f87e 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/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; diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java index 37ee88e2..19a33667 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java @@ -15,18 +15,27 @@ public class ChannelFutureAggregator implements ChannelFutureListener { private ClientRequestResult result; - public ChannelFutureAggregator(String dest, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency) { - this(dest, request, type, amount, latency, 0); + public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, boolean empty) { + this(destination, request, type, amount, latency, empty, (short) 0); } - private ChannelFutureAggregator(String dest, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, int errorCode) { + 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() - .dest(dest) + .destination(destination) .type(type) .request(request) .amount(amount) .latency(latency) .errorCode(errorCode) + .empty(empty) .build(); } @@ -36,7 +45,7 @@ public class ChannelFutureAggregator implements ChannelFutureListener { if (future.getCause() != null) { result.channelError = future.getCause(); } - profiler().profiling(result.dest, result); + profiler().profiling(result.destination, result); } /** @@ -44,39 +53,42 @@ public class ChannelFutureAggregator implements ChannelFutureListener { */ public static class ClientRequestResult { - private String dest; + private String destination; private CanalPacket.PacketType type; private GeneratedMessage request; - private int amount; - private long latency; - private int errorCode; - private Throwable channelError; + private int amount; + private long latency; + private short errorCode; + private boolean empty; + private Throwable channelError; private ClientRequestResult() {} private ClientRequestResult(Builder builder) { - this.dest = Preconditions.checkNotNull(builder.dest); + this.destination = Preconditions.checkNotNull(builder.destination); this.type = Preconditions.checkNotNull(builder.type); this.request = Preconditions.checkNotNull(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 dest; + private String destination; private CanalPacket.PacketType type; private GeneratedMessage request; private int amount; private long latency; - private int errorCode; + private short errorCode; + private boolean empty; private Throwable channelError; - Builder dest(String dest) { - this.dest = dest; + Builder destination(String destination) { + this.destination = destination; return this; } @@ -100,23 +112,29 @@ public class ChannelFutureAggregator implements ChannelFutureListener { return this; } - Builder errorCode(int errorCode) { + 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) { - dest = prototype.dest; + 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; } @@ -126,8 +144,8 @@ public class ChannelFutureAggregator implements ChannelFutureListener { } } // getters - public String getDest() { - return dest; + public String getDestination() { + return destination; } public CanalPacket.PacketType getType() { @@ -146,10 +164,14 @@ public class ChannelFutureAggregator implements ChannelFutureListener { return latency; } - public int getErrorCode() { + public short getErrorCode() { return errorCode; } + public boolean getEmpty() { + return empty; + } + public Throwable getChannelError() { return channelError; } diff --git a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java index 6511e6ba..23c5071b 100644 --- a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java +++ b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java @@ -39,6 +39,7 @@ public class EntryEventSink extends AbstractCanalEventSink> 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> handler : getHandlers()) { events = handler.after(events); } return true; } else { + if (fullTimes == 0) { + blockingStart = System.nanoTime(); + } applyWait(++fullTimes); } @@ -202,4 +209,8 @@ public class EntryEventSink extends AbstractCanalEventSink