canal server metrics refactor.

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<String> 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<String> 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());
+    }
 }

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;
         }
     }

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();
         }

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<String>        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<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        CounterMetricFamily blockingCounter = new CounterMetricFamily(SINK_BLOCKING_TIME,
+                sinkBlockTimeHelp, DEST_LABEL_LIST);
+        blockingCounter.addMetric(destLabelValues, eventsSinkBlockingTime.doubleValue());
+        mfs.add(blockingCounter);
+        return mfs;
+    }
+
+}

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

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

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

@@ -55,7 +55,7 @@ public class InstanceMetaCollector extends Collector {
             List<ClientIdentity> 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);
         }

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<MemoryEventStoreWithBuffer> clazz  = MemoryEventStoreWithBuffer.class;
-
     private final String                                   destination;
-
     private final AtomicLong                               putSequence;
-
     private final AtomicLong                               ackSequence;
-
     private final String                                   putHelp;
-
     private final String                                   ackHelp;
 
     public MemoryStoreCollector(CanalEventStore store, String destination) {
@@ -47,11 +42,11 @@ public class MemoryStoreCollector extends Collector {
     public List<MetricFamilySamples> collect() {
         List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
         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;

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<String>        destLabelValues;
+    private final List<String>        modeLabelValues;
+    private final SerializedLongAdder receivedBinlogBytes;
+    private final SerializedLongAdder eventsPublishBlockingTime;
+    private final String              publishBlockingTimeHelp;
+    private final String              receivedBinlogBytesHelp;
+    private final String              modeHelp;
+    private final List<String>        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<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
+        // 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;
+    }
+}

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

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

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

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;
+    }
+}

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);
+    }
+}

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

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