Merge branch 'master' of https://github.com/lcybo/canal

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

@@ -7,7 +7,6 @@ import com.alibaba.otter.canal.sink.CanalEventSink;
 import com.alibaba.otter.canal.sink.entry.EntryEventSink;
 import com.google.common.base.Preconditions;
 import io.prometheus.client.Collector;
-import io.prometheus.client.Counter;
 import io.prometheus.client.CounterMetricFamily;
 import io.prometheus.client.GaugeMetricFamily;
 import org.slf4j.Logger;
@@ -30,12 +29,8 @@ public class EntryCollector extends Collector implements InstanceRegistry {
     private static final Logger                             logger            = LoggerFactory.getLogger(SinkCollector.class);
     private static final String                             DELAY             = "canal_instance_traffic_delay";
     private static final String                             TRANSACTION       = "canal_instance_transactions";
-    private static final String                             ROW_EVENTS        = "canal_instance_row_events";
-    private static final String                             ROWS_COUNTER      = "canal_instance_rows_counter";
     private static final String                             DELAY_HELP        = "Traffic delay of canal instance in milliseconds";
     private static final String                             TRANSACTION_HELP  = "Transactions counter of canal instance";
-    private static final String                             ROW_EVENTS_HELP   = "Rowdata events counter of canal instance";
-    private static final String                             ROWS_COUNTER_HELP = "Rows counter of canal instance";
     private final ConcurrentMap<String, EntryMetricsHolder> instances        = new ConcurrentHashMap<String, EntryMetricsHolder>();
 
     private EntryCollector() {}
@@ -55,10 +50,6 @@ public class EntryCollector extends Collector implements InstanceRegistry {
                 DELAY_HELP, DEST_LABELS_LIST);
         CounterMetricFamily transactions = new CounterMetricFamily(TRANSACTION,
                 TRANSACTION_HELP, DEST_LABELS_LIST);
-        CounterMetricFamily rowEvents = new CounterMetricFamily(ROW_EVENTS,
-                ROW_EVENTS_HELP, DEST_LABELS_LIST);
-        CounterMetricFamily rowsCounter = new CounterMetricFamily(ROWS_COUNTER,
-                ROWS_COUNTER_HELP, DEST_LABELS_LIST);
         for (EntryMetricsHolder emh : instances.values()) {
             long now = System.currentTimeMillis();
             long latest = emh.latestExecTime.get();
@@ -66,13 +57,9 @@ public class EntryCollector extends Collector implements InstanceRegistry {
                 delay.addMetric(emh.destLabelValues, (now - latest));
             }
             transactions.addMetric(emh.destLabelValues, emh.transactionCounter.doubleValue());
-            rowEvents.addMetric(emh.destLabelValues, emh.rowEventCounter.doubleValue());
-            rowsCounter.addMetric(emh.destLabelValues, emh.rowsCounter.doubleValue());
         }
         mfs.add(delay);
         mfs.add(transactions);
-        mfs.add(rowEvents);
-        mfs.add(rowsCounter);
         return mfs;
     }
 
@@ -89,12 +76,8 @@ public class EntryCollector extends Collector implements InstanceRegistry {
         PrometheusCanalEventDownStreamHandler handler = assembleHandler(entrySink);
         holder.latestExecTime = handler.getLatestExecuteTime();
         holder.transactionCounter = handler.getTransactionCounter();
-        holder.rowEventCounter = handler.getRowEventCounter();
-        holder.rowsCounter = handler.getRowsCounter();
         Preconditions.checkNotNull(holder.latestExecTime);
         Preconditions.checkNotNull(holder.transactionCounter);
-        Preconditions.checkNotNull(holder.rowEventCounter);
-        Preconditions.checkNotNull(holder.rowsCounter);
         EntryMetricsHolder old = instances.put(destination, holder);
         if (old != null) {
             logger.warn("Remove stale EntryCollector for instance {}.", destination);
@@ -145,8 +128,6 @@ public class EntryCollector extends Collector implements InstanceRegistry {
     private class EntryMetricsHolder {
         private AtomicLong   latestExecTime;
         private AtomicLong   transactionCounter;
-        private AtomicLong   rowEventCounter;
-        private AtomicLong   rowsCounter;
         private List<String> destLabelValues;
     }
 

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

@@ -13,10 +13,8 @@ import java.util.concurrent.atomic.AtomicLong;
  */
 public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDownStreamHandler<List<Event>> {
 
-    private final AtomicLong latestExecuteTime  = new AtomicLong(0L);
+    private final AtomicLong latestExecuteTime  = new AtomicLong(System.currentTimeMillis());
     private final AtomicLong transactionCounter = new AtomicLong(0L);
-    private final AtomicLong rowEventCounter    = new AtomicLong(0L);
-    private final AtomicLong rowsCounter        = new AtomicLong(0L);
 
     @Override
     public List<Event> before(List<Event> events) {
@@ -34,8 +32,6 @@ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDow
                     case ROWDATA: {
                         long exec = e.getExecuteTime();
                         if (exec > 0) localExecTime = exec;
-                        rowEventCounter.incrementAndGet();
-                        rowsCounter.addAndGet(e.getRowsCount());
                         break;
                     }
                     case TRANSACTIONEND: {
@@ -80,11 +76,4 @@ public class PrometheusCanalEventDownStreamHandler extends AbstractCanalEventDow
         return transactionCounter;
     }
 
-    public AtomicLong getRowsCounter() {
-        return rowsCounter;
-    }
-
-    public AtomicLong getRowEventCounter() {
-        return rowEventCounter;
-    }
 }

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

@@ -71,7 +71,6 @@ public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler
         String destination = result.getDestination();
         PacketType type = result.getType();
         outboundCounter.labels(destination).inc(result.getAmount());
-        packetsCounter.labels(destination, type.name()).inc();
         short errorCode = result.getErrorCode();
         if (errorCode > 0) {
             errorsCounter.labels(destination, Short.toString(errorCode)).inc();
@@ -81,12 +80,16 @@ public class PrometheusClientInstanceProfiler implements ClientInstanceProfiler
         switch (type) {
             case GET:
                 boolean empty = result.getEmpty();
+                // 区分一下空包
                 if (empty) {
                     emptyBatchesCounter.labels(destination).inc();
+                } else {
+                    packetsCounter.labels(destination, type.name()).inc();
                 }
                 break;
             // reserve for others
             default:
+                packetsCounter.labels(destination, type.name()).inc();
                 break;
         }
     }

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

@@ -34,11 +34,23 @@ public class StoreCollector extends Collector implements InstanceRegistry {
     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                             PUT_DELAY        = "canal_instance_put_delay";
+    private static final String                             GET_DELAY        = "canal_instance_get_delay";
+    private static final String                             ACK_DELAY        = "canal_instance_ack_delay";
+    private static final String                             PUT_ROWS         = "canal_instance_put_rows";
+    private static final String                             GET_ROWS         = "canal_instance_get_rows";
+    private static final String                             ACK_ROWS         = "canal_instance_ack_rows";
     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 static final String                             PUT_DELAY_HELP   = "Traffic delay of canal instance put";
+    private static final String                             GET_DELAY_HELP   = "Traffic delay of canal instance get";
+    private static final String                             ACK_DELAY_HELP   = "Traffic delay of canal instance ack";
+    private static final String                             PUT_ROWS_HELP    = "Put table rows of canal instance";
+    private static final String                             GET_ROWS_HELP    = "Got table rows of canal instance";
+    private static final String                             ACK_ROWS_HELP    = "Acked table rows of canal instance";
     private final ConcurrentMap<String, StoreMetricsHolder> instances        = new ConcurrentHashMap<String, StoreMetricsHolder>();
     private final List<String>                              storeLabelsList  = Arrays.asList(DEST, "batchMode", "size");
 
@@ -65,11 +77,40 @@ public class StoreCollector extends Collector implements InstanceRegistry {
                 PRODUCE_MEM_HELP, DEST_LABELS_LIST);
         CounterMetricFamily ackMem = new CounterMetricFamily(CONSUME_MEM,
                 CONSUME_MEM_HELP, DEST_LABELS_LIST);
+        GaugeMetricFamily putDelay = new GaugeMetricFamily(PUT_DELAY,
+                PUT_DELAY_HELP, DEST_LABELS_LIST);
+        GaugeMetricFamily getDelay = new GaugeMetricFamily(GET_DELAY,
+                GET_DELAY_HELP, DEST_LABELS_LIST);
+        GaugeMetricFamily ackDelay = new GaugeMetricFamily(ACK_DELAY,
+                ACK_DELAY_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily putRows = new CounterMetricFamily(PUT_ROWS,
+                PUT_ROWS_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily getRows = new CounterMetricFamily(GET_ROWS,
+                GET_ROWS_HELP, DEST_LABELS_LIST);
+        CounterMetricFamily ackRows = new CounterMetricFamily(ACK_ROWS,
+                ACK_ROWS_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());
+            long pet = smh.putExecTime.get();
+            // 防止出现启动时，未消费造成的get, ack延时小于前阶段的情况
+            long get = Math.min(smh.getExecTime.get(), pet);
+            long aet = Math.min(smh.ackExecTime.get(), get);
+            long now = System.currentTimeMillis();
+            if (now >= pet) {
+                putDelay.addMetric(smh.destLabelValues, (now - pet));
+            }
+            if (now >= get) {
+                getDelay.addMetric(smh.destLabelValues, (now - get));
+            }
+            if (now >= aet) {
+                ackDelay.addMetric(smh.destLabelValues, (now - aet));
+            }
+            putRows.addMetric(smh.destLabelValues, smh.putTableRows.doubleValue());
+            getRows.addMetric(smh.destLabelValues, smh.getTableRows.doubleValue());
+            ackRows.addMetric(smh.destLabelValues, smh.ackTableRows.doubleValue());
             store.addMetric(smh.storeLabelValues, 1);
             if (isMem) {
                 hasMem = true;
@@ -80,6 +121,12 @@ public class StoreCollector extends Collector implements InstanceRegistry {
         mfs.add(put);
         mfs.add(ack);
         mfs.add(store);
+        mfs.add(putDelay);
+        mfs.add(getDelay);
+        mfs.add(ackDelay);
+        mfs.add(putRows);
+        mfs.add(getRows);
+        mfs.add(ackRows);
         if (hasMem) {
             mfs.add(putMem);
             mfs.add(ackMem);
@@ -102,6 +149,12 @@ public class StoreCollector extends Collector implements InstanceRegistry {
         holder.destLabelValues = Collections.singletonList(destination);
         holder.size = memStore.getBufferSize();
         holder.storeLabelValues = Arrays.asList(destination, holder.batchMode.name(), Integer.toString(holder.size));
+        holder.putExecTime = memStore.getPutExecTime();
+        holder.getExecTime = memStore.getGetExecTime();
+        holder.ackExecTime = memStore.getAckExecTime();
+        holder.putTableRows = memStore.getPutTableRows();
+        holder.getTableRows = memStore.getGetTableRows();
+        holder.ackTableRows = memStore.getAckTableRows();
         Preconditions.checkNotNull(holder.batchMode);
         Preconditions.checkNotNull(holder.putSeq);
         Preconditions.checkNotNull(holder.ackSeq);
@@ -129,6 +182,12 @@ public class StoreCollector extends Collector implements InstanceRegistry {
         private BatchMode    batchMode;
         private AtomicLong   putMemSize;
         private AtomicLong   ackMemSize;
+        private AtomicLong   putExecTime;
+        private AtomicLong   getExecTime;
+        private AtomicLong   ackExecTime;
+        private AtomicLong   putTableRows;
+        private AtomicLong   getTableRows;
+        private AtomicLong   ackTableRows;
         private int          size;
         private List<String> destLabelValues;
         private List<String> storeLabelValues;

server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java

@@ -220,6 +220,7 @@ public class SessionHandler extends SimpleChannelHandler {
                         } else {
                             clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId()));
                             embeddedServer.ack(clientIdentity, ack.getBatchId());
+                            new ChannelFutureAggregator(ack.getDestination(), ack, packet.getType(), 0, System.nanoTime() - start).operationComplete(null);
                         }
                     } else {
                         byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage());
@@ -239,6 +240,7 @@ public class SessionHandler extends SimpleChannelHandler {
                         } else {
                             embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
                         }
+                        new ChannelFutureAggregator(rollback.getDestination(), rollback, packet.getType(), 0, System.nanoTime() - start).operationComplete(null);
                     } else {
                         byte[] errorBytes = NettyUtils.errorPacket(401, MessageFormatter.format("destination or clientId is null", rollback.toString()).getMessage());
                         NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(rollback.getDestination(),

server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java

@@ -43,7 +43,7 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
     @Override
     public void operationComplete(ChannelFuture future) {
         // profiling after I/O operation
-        if (future.getCause() != null) {
+        if (future != null && future.getCause() != null) {
             result.channelError = future.getCause();
         }
         profiler().profiling(result);

store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java

@@ -51,6 +51,16 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
     private AtomicLong        getMemSize    = new AtomicLong(0);
     private AtomicLong        ackMemSize    = new AtomicLong(0);
 
+    // 记录下put/get/ack操作的三个execTime
+    private AtomicLong        putExecTime   = new AtomicLong(System.currentTimeMillis());
+    private AtomicLong        getExecTime   = new AtomicLong(System.currentTimeMillis());
+    private AtomicLong        ackExecTime   = new AtomicLong(System.currentTimeMillis());
+
+    // 记录下put/get/ack操作的三个table rows
+    private AtomicLong        putTableRows  = new AtomicLong(0);
+    private AtomicLong        getTableRows  = new AtomicLong(0);
+    private AtomicLong        ackTableRows  = new AtomicLong(0);
+
     // 阻塞put/get操作控制信号
     private ReentrantLock     lock          = new ReentrantLock();
     private Condition         notFull       = lock.newCondition();
@@ -192,7 +202,7 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
 
             putMemSize.getAndAdd(size);
         }
-
+        profiling(data, OP.PUT);
         // tell other threads that store is not empty
         notEmpty.signal();
     }
@@ -336,6 +346,7 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
         if (getSequence.compareAndSet(current, end)) {
             getMemSize.addAndGet(memsize);
             notFull.signal();
+            profiling(result.getEvents(), OP.GET);
             return result;
         } else {
             return new Events<Event>();
@@ -407,8 +418,15 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
 
             boolean hasMatch = false;
             long memsize = 0;
+            // ack没有list，但有已存在的foreach，还是节省一下list的开销
+            long localExecTime = 0L;
+            int deltaRows = 0;
             for (long next = sequence + 1; next <= maxSequence; next++) {
                 Event event = entries[getIndex(next)];
+                if (localExecTime == 0 && event.getExecuteTime() > 0) {
+                    localExecTime = event.getExecuteTime();
+                }
+                deltaRows += event.getRowsCount();
                 memsize += calculateSize(event);
                 boolean match = CanalEventUtils.checkPosition(event, (LogPosition) position);
                 if (match) {// 找到对应的position，更新ack seq
@@ -424,11 +442,14 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
 
                     if (ackSequence.compareAndSet(sequence, next)) {// 避免并发ack
                         notFull.signal();
+                        ackTableRows.addAndGet(deltaRows);
+                        if (localExecTime > 0) {
+                            ackExecTime.lazySet(localExecTime);
+                        }
                         return;
                     }
                 }
             }
-
             if (!hasMatch) {// 找不到对应需要ack的position
                 throw new CanalStoreException("no match ack position" + position.toString());
             }
@@ -544,6 +565,45 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
                || type == EventType.DINDEX;
     }
 
+    private void profiling(List<Event> events, OP op) {
+        long localExecTime = 0L;
+        int deltaRows = 0;
+        if (events != null && !events.isEmpty()) {
+            for (Event e : events) {
+                if (localExecTime == 0 && e.getExecuteTime() > 0) {
+                    localExecTime = e.getExecuteTime();
+                }
+                deltaRows += e.getRowsCount();
+            }
+        }
+        switch (op) {
+            case PUT:
+                putTableRows.addAndGet(deltaRows);
+                if (localExecTime > 0) {
+                    putExecTime.lazySet(localExecTime);
+                }
+                break;
+            case GET:
+                getTableRows.addAndGet(deltaRows);
+                if (localExecTime > 0) {
+                    getExecTime.lazySet(localExecTime);
+                }
+                break;
+            case ACK:
+                ackTableRows.addAndGet(deltaRows);
+                if (localExecTime > 0) {
+                    ackExecTime.lazySet(localExecTime);
+                }
+                break;
+            default:
+                break;
+        }
+    }
+
+    private enum OP {
+        PUT, GET, ACK
+    }
+
     // ================ setter / getter ==================
     public int getBufferSize() {
         return this.bufferSize;
@@ -584,4 +644,28 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple
     public BatchMode getBatchMode() {
         return batchMode;
     }
+
+    public AtomicLong getPutExecTime() {
+        return putExecTime;
+    }
+
+    public AtomicLong getGetExecTime() {
+        return getExecTime;
+    }
+
+    public AtomicLong getAckExecTime() {
+        return ackExecTime;
+    }
+
+    public AtomicLong getPutTableRows() {
+        return putTableRows;
+    }
+
+    public AtomicLong getGetTableRows() {
+        return getTableRows;
+    }
+
+    public AtomicLong getAckTableRows() {
+        return ackTableRows;
+    }
 }