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 deleted file mode 100644 index 4d86a99e..00000000 --- a/common/src/main/java/com/alibaba/otter/canal/common/utils/SerializedLongAdder.java +++ /dev/null @@ -1,31 +0,0 @@ -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 ff92dd9d..a83c3516 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 @@ -8,7 +8,6 @@ import java.util.TimerTask; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -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; 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 5e3c6b02..156d4692 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 @@ -1,8 +1,8 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.nio.charset.Charset; +import java.util.concurrent.atomic.AtomicLong; -import com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,7 +35,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); + private final AtomicLong eventsPublishBlockingTime = new AtomicLong(0L); protected BinlogParser buildParser() { LogEventConvert convert = new LogEventConvert(); @@ -192,7 +192,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { } } - public SerializedLongAdder getEventsPublishBlockingTime() { + public AtomicLong 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 c125dbd3..9d89a145 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 @@ -7,8 +7,8 @@ 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 com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,7 +50,7 @@ public class MysqlConnection implements ErosaConnection { private AuthenticationInfo authInfo; protected int connTimeout = 5 * 1000; // 5秒 protected int soTimeout = 60 * 60 * 1000; // 1小时 - private SerializedLongAdder receivedBinlogBytes; + private AtomicLong receivedBinlogBytes; public MysqlConnection(){ } @@ -461,7 +461,7 @@ public class MysqlConnection implements ErosaConnection { private void accumulateReceivedBytes(long x) { if (receivedBinlogBytes != null) { - receivedBinlogBytes.add(x); + receivedBinlogBytes.addAndGet(x); } } @@ -606,7 +606,7 @@ public class MysqlConnection implements ErosaConnection { this.authInfo = authInfo; } - public void setReceivedBinlogBytes(SerializedLongAdder receivedBinlogBytes) { + public void setReceivedBinlogBytes(AtomicLong 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 a87be542..69cf0572 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,7 +10,6 @@ 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; @@ -74,7 +73,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 // instance received binlog bytes - private final SerializedLongAdder receivedBinlogBytes = new SerializedLongAdder(0L); + private final AtomicLong receivedBinlogBytes = new AtomicLong(0L); protected ErosaConnection buildErosaConnection() { return buildMysqlConnection(this.runningInfo); @@ -517,7 +516,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() { @@ -918,7 +917,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE - public SerializedLongAdder getReceivedBinlogBytes() { + public AtomicLong 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 e3206e3e..92447d50 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 @@ -2,9 +2,9 @@ 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 com.alibaba.otter.canal.common.utils.SerializedLongAdder; import org.apache.commons.lang.StringUtils; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; @@ -65,7 +65,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private ExecutorService stageExecutor; private String destination; private volatile CanalParseException exception; - private SerializedLongAdder eventsPublishBlockingTime; + private AtomicLong eventsPublishBlockingTime; public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, EventTransactionBuffer transactionBuffer, String destination){ @@ -176,7 +176,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement } disruptorMsgBuffer.publish(next); if (fullTimes > 0) { - eventsPublishBlockingTime.add(System.nanoTime() - blockingStart); + eventsPublishBlockingTime.addAndGet(System.nanoTime() - blockingStart); } break; } catch (InsufficientCapacityException e) { @@ -451,7 +451,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement this.connection = connection; } - public void setEventsPublishBlockingTime(SerializedLongAdder eventsPublishBlockingTime) { + public void setEventsPublishBlockingTime(AtomicLong 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 6908a3bd..487855fa 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 @@ -2,9 +2,6 @@ package com.alibaba.otter.canal.prometheus; import com.alibaba.otter.canal.instance.core.CanalInstance; 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; import io.prometheus.client.Collector; import io.prometheus.client.CollectorRegistry; import org.slf4j.Logger; @@ -18,98 +15,71 @@ import java.util.List; */ public class CanalInstanceExports { - 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 DEST_LABEL_LIST = Collections.singletonList(DEST); -// private final String destination; - private Collector storeCollector; - private Collector delayCollector; - private Collector metaCollector; - private Collector sinkCollector; - private Collector parserCollector; + 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 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 CanalInstanceExports() { -// this.destination = instance.getDestination(); -// initEventsMetrics(instance); -// initStoreCollector(instance); -// initMetaCollector(instance); -// initSinkCollector(instance); -// initParserCollector(instance); + private CanalInstanceExports() { + this.storeCollector = StoreCollector.instance(); + this.entryCollector = EntryCollector.instance(); + this.metaCollector = MetaCollector.instance(); + this.sinkCollector = SinkCollector.instance(); + this.parserCollector = ParserCollector.instance(); } + private static class SingletonHolder { + private static final CanalInstanceExports SINGLETON = new CanalInstanceExports(); + } + public static CanalInstanceExports instance() { + return SingletonHolder.SINGLETON; + } -// static CanalInstanceExports forInstance(CanalInstance instance) { -// return new CanalInstanceExports(instance); -// } + public void initialize() { + storeCollector.register(); + entryCollector.register(); + metaCollector.register(); + sinkCollector.register(); + parserCollector.register(); + } + + public void terminate() { + CollectorRegistry.defaultRegistry.unregister(storeCollector); + CollectorRegistry.defaultRegistry.unregister(entryCollector); + CollectorRegistry.defaultRegistry.unregister(metaCollector); + CollectorRegistry.defaultRegistry.unregister(sinkCollector); + CollectorRegistry.defaultRegistry.unregister(parserCollector); + } void register(CanalInstance instance) { - if (delayCollector != null) { - delayCollector.register(); - } - if (storeCollector != null) { - storeCollector.register(); - } - if (metaCollector != null) { - metaCollector.register(); - } - if (sinkCollector != null) { - sinkCollector.register(); - } - if (parserCollector != null) { - parserCollector.register(); - } + 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()); } void unregister(CanalInstance instance) { - if (delayCollector != null) { - CollectorRegistry.defaultRegistry.unregister(delayCollector); - } - if (storeCollector != null) { - CollectorRegistry.defaultRegistry.unregister(storeCollector); - } - if (metaCollector != null) { - CollectorRegistry.defaultRegistry.unregister(metaCollector); - } - if (sinkCollector != null) { - CollectorRegistry.defaultRegistry.unregister(sinkCollector); - } - if (parserCollector != null) { - CollectorRegistry.defaultRegistry.unregister(parserCollector); - } + 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 void initEventsMetrics(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."); + private InstanceRegistry requiredInstanceRegistry(Collector collector) { + if (!(collector instanceof InstanceRegistry)) { + throw new IllegalArgumentException("Canal collector need to implement InstanceRegistry."); } + return (InstanceRegistry) collector; } -// private void initStoreCollector(CanalInstance instance) { -// try { -// storeCollector = new StoreCollector(instance.getEventStore(), destination); -// } catch (CanalStoreException cse) { -// logger.warn("Failed to register metrics for destination {}.", destination, cse); -// } -// } - - private void initMetaCollector(CanalInstance instance) { - metaCollector = new MetaCollector(instance); - } - - private void initSinkCollector(CanalInstance instance) { -// sinkCollector = new EntryCollector(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 deleted file mode 100644 index ded44fb8..00000000 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/CanalServerExports.java +++ /dev/null @@ -1,17 +0,0 @@ -package com.alibaba.otter.canal.prometheus; - - -/** - * @author Chuanyi Li - */ -public class CanalServerExports { - - private static boolean initialized = false; - - public static synchronized void initialize() { - if (!initialized) { - 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 c3a241bd..9e3ef18c 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 @@ -20,12 +20,13 @@ import static com.alibaba.otter.canal.server.netty.CanalServerWithNettyProfiler. public class PrometheusService implements CanalMetricsService { private static final Logger logger = LoggerFactory.getLogger(PrometheusService.class); - private final CanalInstanceExports instanceExports = new CanalInstanceExports(); + private final CanalInstanceExports instanceExports; private volatile boolean running = false; private HTTPServer server; private ClientInstanceProfiler clientProfiler = new PrometheusClientInstanceProfiler(); private PrometheusService() { + this.instanceExports = CanalInstanceExports.instance(); } private static class SingletonHolder { @@ -49,8 +50,7 @@ public class PrometheusService implements CanalMetricsService { try { // JVM exports DefaultExports.initialize(); - // Canal server level exports - CanalServerExports.initialize(); + instanceExports.initialize(); if (!clientProfiler.isStart()) { clientProfiler.start(); } @@ -65,13 +65,17 @@ public class PrometheusService implements CanalMetricsService { @Override public void terminate() { running = false; - instanceExports.unregister(); - if (clientProfiler.isStart()) { - clientProfiler.stop(); - } - profiler().setInstanceProfiler(NOP); - 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); } } @@ -87,9 +91,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); } @@ -102,10 +104,7 @@ 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); } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java index 83e4ef95..dd282065 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/EntryCollector.java @@ -2,10 +2,12 @@ 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.CounterMetricFamily; import io.prometheus.client.GaugeMetricFamily; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -17,27 +19,43 @@ 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_LABEL_LIST; +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_NAME = "canal_instance_traffic_delay"; - private static final String delayHelpName = "Traffic delay of canal instance"; - private final ConcurrentMap instances = new ConcurrentHashMap(); + 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"; + private static final String TRANSACTION_HELP = "Transactions counter of canal instance"; + private final ConcurrentMap instances = new ConcurrentHashMap(); + + private EntryCollector() {} + + private static class SingletonHolder { + private static final EntryCollector SINGLETON = new EntryCollector(); + } + + public static EntryCollector instance() { + return SingletonHolder.SINGLETON; + } @Override public List collect() { List mfs = new ArrayList(); - GaugeMetricFamily delay = new GaugeMetricFamily(DELAY_NAME, - delayHelpName, DEST_LABEL_LIST); + GaugeMetricFamily delay = new GaugeMetricFamily(DELAY, + DELAY_HELP, DEST_LABELS_LIST); + CounterMetricFamily transactions = new CounterMetricFamily(TRANSACTION, + TRANSACTION_HELP, DEST_LABELS_LIST); for (EntryMetricsHolder emh : instances.values()) { delay.addMetric(emh.destLabelValues, emh.latestExecTime.doubleValue()); + transactions.addMetric(emh.destLabelValues, emh.transactionCounter.doubleValue()); } mfs.add(delay); + mfs.add(transactions); return mfs; } @@ -51,25 +69,61 @@ public class EntryCollector extends Collector implements InstanceRegistry { throw new IllegalArgumentException("CanalEventSink must be EntryEventSink"); } EntryEventSink entrySink = (EntryEventSink) sink; - PrometheusCanalEventDownStreamHandler handler = new PrometheusCanalEventDownStreamHandler(); + PrometheusCanalEventDownStreamHandler handler = assembleHandler(entrySink); holder.latestExecTime = handler.getLatestExecuteTime(); - entrySink.addHandler(handler, 0); - Preconditions.checkNotNull(holder.destLabelValues); + holder.transactionCounter = handler.getTransactionCounter(); Preconditions.checkNotNull(holder.latestExecTime); - EntryMetricsHolder old = instances.putIfAbsent(destination, holder); + Preconditions.checkNotNull(holder.transactionCounter); + EntryMetricsHolder old = instances.put(destination, holder); if (old != null) { - logger.warn("Ignore repeated EntryCollector register for instance {}.", destination); + 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 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 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 List destLabelValues; } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MetaCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MetaCollector.java index eb5a2bde..aec9a424 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MetaCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MetaCollector.java @@ -23,18 +23,28 @@ import java.util.concurrent.ConcurrentMap; */ public class MetaCollector extends Collector implements InstanceRegistry { - private static final List InfoLabel = Arrays.asList("destination", "mode"); - private static final Logger logger = LoggerFactory.getLogger(MetaCollector.class); - private static final String SUBSCRIPTION = "canal_instance"; - private static final String subscriptionHelp = "Canal instance"; - private final ConcurrentMap instances = new ConcurrentHashMap(); + private static final List InfoLabel = Arrays.asList("destination", "mode"); + private static final Logger logger = LoggerFactory.getLogger(MetaCollector.class); + private static final String SUBSCRIPTION = "canal_instance"; + private static final String SUBSCRIPTION_HELP = "Canal instance"; + private final ConcurrentMap instances = new ConcurrentHashMap(); + + private MetaCollector() {} + + private static class SingletonHolder { + private static final MetaCollector SINGLETON = new MetaCollector(); + } + + public static MetaCollector instance() { + return SingletonHolder.SINGLETON; + } @Override public List collect() { List mfs = new ArrayList(); GaugeMetricFamily instanceInfo = new GaugeMetricFamily( SUBSCRIPTION, - subscriptionHelp, + SUBSCRIPTION_HELP, InfoLabel); for (Map.Entry nme : instances.entrySet()) { final String destination = nme.getKey(); @@ -54,11 +64,10 @@ public class MetaCollector extends Collector implements InstanceRegistry { String mode = (instance instanceof CanalInstanceWithSpring) ? "spring" : "manager"; holder.infoLabelValues = Arrays.asList(destination, mode); holder.metaManager = instance.getMetaManager(); - Preconditions.checkNotNull(holder.infoLabelValues); Preconditions.checkNotNull(holder.metaManager); - MetaMetricsHolder old = instances.putIfAbsent(destination, holder); + MetaMetricsHolder old = instances.put(destination, holder); if (old != null) { - logger.warn("Ignore repeated MetaCollector register for instance {}.", destination); + logger.warn("Remove stale MetaCollector for instance {}.", destination); } } 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 deleted file mode 100644 index c8543f3c..00000000 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/MysqlParserCollector.java +++ /dev/null @@ -1,74 +0,0 @@ -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.DEST; -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(DEST, 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/ParserCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java new file mode 100644 index 00000000..60906989 --- /dev/null +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/ParserCollector.java @@ -0,0 +1,114 @@ +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; + +/** + * @author Chuanyi Li + */ +public class ParserCollector extends Collector implements InstanceRegistry { + + private static final Logger logger = LoggerFactory.getLogger(ParserCollector.class); + 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"; + 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 modeLabels = Arrays.asList(DEST, MODE_LABEL); + private final ConcurrentMap instances = new ConcurrentHashMap(); + + private ParserCollector() {} + + private static class SingletonHolder { + private static final ParserCollector SINGLETON = new ParserCollector(); + } + + public static ParserCollector instance() { + return SingletonHolder.SINGLETON; + } + + @Override + public List collect() { + List mfs = new ArrayList(); + 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()); + 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 destLabelValues; + private List modeLabelValues; + private AtomicLong receivedBinlogBytes; + private AtomicLong eventsPublishBlockingTime; + private boolean isParallel; + } + +} 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 763ce46b..6fc2f3d7 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 @@ -1,7 +1,9 @@ package com.alibaba.otter.canal.prometheus.impl; +import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import com.alibaba.otter.canal.sink.AbstractCanalEventDownStreamHandler; import com.alibaba.otter.canal.store.model.Event; + import java.util.List; import java.util.concurrent.atomic.AtomicLong; @@ -10,28 +12,71 @@ import java.util.concurrent.atomic.AtomicLong; */ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDownStreamHandler> { - private final AtomicLong latestExecuteTime = new AtomicLong(0L); + 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 before(List 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(); - long ls = latestExecuteTime.get(); - if (ts > ls) { - latestExecuteTime.lazySet(ts); + for (Event e : events) { + EntryType type = e.getEntryType(); + if (type == null) continue; + long exec = e.getExecuteTime(); + if (exec > 0) localExecTime = exec; + switch (type) { + case ROWDATA: + // TODO 当前proto无法直接获得荣威change的变更行数（需要parse），可考虑放到header里面 + break; + case TRANSACTIONEND: + transactionCounter.incrementAndGet(); + break; + case HEARTBEAT: + // TODO 确认一下不是canal自己产生的 + // 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. + break; + default: + break; + } + } + if (localExecTime > 0) { + latestExecuteTime.lazySet(localExecTime); } } return events; } + @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; + } } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java index 11903314..fc5ed3b2 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/SinkCollector.java @@ -17,23 +17,33 @@ 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_LABEL_LIST; +import static com.alibaba.otter.canal.prometheus.CanalInstanceExports.DEST_LABELS_LIST; /** * @author Chuanyi Li */ public class SinkCollector extends Collector implements InstanceRegistry { - private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class); - private static final String SINK_BLOCKING_TIME = "canal_instance_sink_blocking_time"; - private static final String sinkBlockTimeHelp = "Total sink blocking time"; - private final ConcurrentMap instances = new ConcurrentHashMap(); + private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class); + private static final String SINK_BLOCKING_TIME = "canal_instance_sink_blocking_time"; + private static final String SINK_BLOCK_TIME_HELP = "Total sink blocking time"; + private final ConcurrentMap instances = new ConcurrentHashMap(); + + private SinkCollector() {} + + private static class SingletonHolder { + private static final SinkCollector SINGLETON = new SinkCollector(); + } + + public static SinkCollector instance() { + return SingletonHolder.SINGLETON; + } @Override public List collect() { List mfs = new ArrayList(); CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME, - sinkBlockTimeHelp, DEST_LABEL_LIST); + SINK_BLOCK_TIME_HELP, DEST_LABELS_LIST); for (SinkMetricsHolder smh : instances.values()) { blockingCounter.addMetric(smh.destLabelValues, smh.eventsSinkBlockingTime.doubleValue()); } @@ -52,11 +62,10 @@ public class SinkCollector extends Collector implements InstanceRegistry { } EntryEventSink entrySink = (EntryEventSink) sink; holder.eventsSinkBlockingTime = entrySink.getEventsSinkBlockingTime(); - Preconditions.checkNotNull(holder.destLabelValues); Preconditions.checkNotNull(holder.eventsSinkBlockingTime); - SinkMetricsHolder old = instances.putIfAbsent(destination, holder); + SinkMetricsHolder old = instances.put(destination, holder); if (old != null) { - logger.warn("Ignore repeated SinkCollector register for instance {}.", destination); + logger.warn("Remote stale SinkCollector for instance {}.", destination); } } diff --git a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java index 3784af9b..8bd39574 100644 --- a/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java +++ b/prometheus/src/main/java/com/alibaba/otter/canal/prometheus/impl/StoreCollector.java @@ -1,15 +1,17 @@ package com.alibaba.otter.canal.prometheus.impl; import com.alibaba.otter.canal.instance.core.CanalInstance; -import com.alibaba.otter.canal.prometheus.CanalInstanceExports; import com.alibaba.otter.canal.prometheus.InstanceRegistry; import com.alibaba.otter.canal.store.CanalEventStore; -import com.alibaba.otter.canal.store.CanalStoreException; 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.lang.reflect.Field; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -18,68 +20,113 @@ 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 Class clazz = MemoryEventStoreWithBuffer.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 produceHelp = "Produced sequence of canal instance"; - private static final String consumeHelp = "Consumed sequence of canal instance"; - private final ConcurrentMap instances = new ConcurrentHashMap(); + 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 instances = new ConcurrentHashMap(); + private final List storeLabelsList = Arrays.asList(DEST, "batchMode"); + private StoreCollector() {} - public StoreCollector(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"); + private static class SingletonHolder { + private static final StoreCollector SINGLETON = new StoreCollector(); + } + + public static StoreCollector instance() { + return SingletonHolder.SINGLETON; } @Override public List collect() { List mfs = new ArrayList(); - CounterMetricFamily put = new CounterMetricFamily("canal_instance_store_produce_seq", - putHelp, Arrays.asList(CanalInstanceExports.DEST_LABELS)); - put.addMetric(Collections.singletonList(destination), putSequence.doubleValue()); + 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); - CounterMetricFamily ack = new CounterMetricFamily("canal_instance_store_consume_seq", - ackHelp, Arrays.asList(CanalInstanceExports.DEST_LABELS)); - ack.addMetric(Collections.singletonList(destination), ackSequence.doubleValue()); mfs.add(ack); + mfs.add(store); + if (hasMem) { + mfs.add(putMem); + mfs.add(ackMem); + } return mfs; } - @SuppressWarnings("unchecked") - private static 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; - } - @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 AtomicLong putSeq; + private AtomicLong ackSeq; + private BatchMode batchMode; + private AtomicLong putMemSize; + private AtomicLong ackMemSize; + private List destLabelValues; + private List storeLabelValues; } } diff --git a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java index 075d1524..00ba2e71 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java +++ b/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; + } }