rds binlog 下载支持

server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java

@@ -1,11 +1,11 @@
 package com.alibaba.otter.canal.server.embedded;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
+import com.alibaba.otter.canal.spi.CanalMetricsProvider;
+import com.alibaba.otter.canal.spi.CanalMetricsService;
+import com.alibaba.otter.canal.spi.NopCanalMetricsService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
@@ -33,17 +33,18 @@ import com.google.protobuf.ByteString;
 
 /**
  * 嵌入式版本实现
- * 
+ *
  * @author jianghang 2012-7-12 下午01:34:00
  * @author zebin.xuzb
  * @version 1.0.0
  */
 public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService {
 
-    private static final Logger        logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
+    private static final Logger        logger           = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
     private Map<String, CanalInstance> canalInstances;
     // private Map<ClientIdentity, Position> lastRollbackPostions;
     private CanalInstanceGenerator     canalInstanceGenerator;
+    private CanalMetricsService        metrics          = NopCanalMetricsService.NOP;
 
     private static class SingletonHolder {
 
@@ -61,7 +62,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
     public void start() {
         if (!isStart()) {
             super.start();
-
+            // 如果存在provider,则启动metrics service
+            loadCanalMetrics();
+            metrics.initialize();
             canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>() {
 
                 public CanalInstance apply(String destination) {
@@ -92,6 +95,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
                 logger.error(String.format("stop CanalInstance[%s] has an error", entry.getKey()), e);
             }
         }
+        metrics.terminate();
     }
 
     public void start(final String destination) {
@@ -99,6 +103,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
         if (!canalInstance.isStart()) {
             try {
                 MDC.put("destination", destination);
+                if (metrics.isRunning()) {
+                    metrics.register(canalInstance);
+                }
                 canalInstance.start();
                 logger.info("start CanalInstances[{}] successfully", destination);
             } finally {
@@ -114,6 +121,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
                 try {
                     MDC.put("destination", destination);
                     canalInstance.stop();
+                    if (metrics.isRunning()) {
+                        metrics.unregister(canalInstance);
+                    }
                     logger.info("stop CanalInstances[{}] successfully", destination);
                 } finally {
                     MDC.remove("destination");
@@ -176,7 +186,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
 
     /**
      * 获取数据
-     * 
+     *
      * <pre>
      * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
      * </pre>
@@ -188,14 +198,14 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
 
     /**
      * 获取数据，可以指定超时时间.
-     * 
+     *
      * <pre>
      * 几种case:
      * a. 如果timeout为null，则采用tryGet方式，即时获取
      * b. 如果timeout不为null
      *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
      *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
-     * 
+     *
      * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
      * </pre>
      */
@@ -251,7 +261,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
     /**
      * 不指定 position 获取事件。canal 会记住此 client 最新的 position。 <br/>
      * 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
-     * 
+     *
      * <pre>
      * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
      * </pre>
@@ -264,14 +274,14 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
     /**
      * 不指定 position 获取事件。canal 会记住此 client 最新的 position。 <br/>
      * 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
-     * 
+     *
      * <pre>
      * 几种case:
      * a. 如果timeout为null，则采用tryGet方式，即时获取
      * b. 如果timeout不为null
      *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
      *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
-     *    
+     *
      * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
      * </pre>
      */
@@ -342,7 +352,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
 
     /**
      * 进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
-     * 
+     *
      * <pre>
      * 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证)
      * </pre>
@@ -492,6 +502,27 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
         }
     }
 
+    private void loadCanalMetrics() {
+        ServiceLoader<CanalMetricsProvider> providers = ServiceLoader.load(CanalMetricsProvider.class);
+        List<CanalMetricsProvider> list = new ArrayList<CanalMetricsProvider>();
+        for (CanalMetricsProvider provider : providers) {
+            list.add(provider);
+        }
+        if (!list.isEmpty()) {
+            // 发现provider, 进行初始化
+            if (list.size() > 1) {
+                logger.warn("Found more than one CanalMetricsProvider, use the first one.");
+                //报告冲突
+                for (CanalMetricsProvider p : list) {
+                    logger.warn("Found CanalMetricsProvider: {}.", p.getClass().getName());
+                }
+            }
+            //默认使用第一个
+            CanalMetricsProvider provider = list.get(0);
+            this.metrics = provider.getService();
+        }
+    }
+
     // ========= setter ==========
 
     public void setCanalInstanceGenerator(CanalInstanceGenerator canalInstanceGenerator) {

server/src/main/java/com/alibaba/otter/canal/server/netty/NettyUtils.java

@@ -2,8 +2,12 @@ package com.alibaba.otter.canal.server.netty;
 
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.List;
 
+import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.buffer.CompositeChannelBuffer;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFutureListener;
 import org.jboss.netty.channel.Channels;
@@ -19,9 +23,23 @@ import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
 public class NettyUtils {
 
     private static final Logger logger           = LoggerFactory.getLogger(NettyUtils.class);
-    private static int          HEADER_LENGTH    = 4;
+    public static int           HEADER_LENGTH    = 4;
     public static Timer         hashedWheelTimer = new HashedWheelTimer();
 
+    public static void write(Channel channel, ByteBuffer body, ChannelFutureListener channelFutureListner) {
+        byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(body.limit()).array();
+        List<ChannelBuffer> components = new ArrayList<ChannelBuffer>(2);
+        components.add(ChannelBuffers.wrappedBuffer(ByteOrder.BIG_ENDIAN, header));
+        components.add(ChannelBuffers.wrappedBuffer(body));
+
+        if (channelFutureListner == null) {
+            Channels.write(channel, new CompositeChannelBuffer(ByteOrder.BIG_ENDIAN, components));
+        } else {
+            Channels.write(channel, new CompositeChannelBuffer(ByteOrder.BIG_ENDIAN, components))
+                .addListener(channelFutureListner);
+        }
+    }
+
     public static void write(Channel channel, byte[] body, ChannelFutureListener channelFutureListner) {
         byte[] header = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN).putInt(body.length).array();
         if (channelFutureListner == null) {

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

@@ -33,13 +33,10 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
 import com.alibaba.otter.canal.server.netty.NettyUtils;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.CodedOutputStream;
+import com.google.protobuf.WireFormat;
 
-/**
- * 处理具体的客户端请求
- * 
- * @author jianghang 2012-10-24 下午02:21:13
- * @version 1.0.0
- */
 public class SessionHandler extends SimpleChannelHandler {
 
     private static final Logger     logger = LoggerFactory.getLogger(SessionHandler.class);
@@ -76,7 +73,7 @@ public class SessionHandler extends SimpleChannelHandler {
                         }
 
                         embeddedServer.subscribe(clientIdentity);
-                        ctx.setAttachment(clientIdentity);// 设置状态数据
+                        // ctx.setAttachment(clientIdentity);// 设置状态数据
                         NettyUtils.ack(ctx.getChannel(), null);
                     } else {
                         NettyUtils.error(401,
@@ -130,27 +127,74 @@ public class SessionHandler extends SimpleChannelHandler {
                         }
                         // }
 
-                        Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
-                        packetBuilder.setType(PacketType.MESSAGES);
+                        if (message.getId() != -1 && message.isRaw()) {
+                            List<ByteString> rowEntries = message.getRawEntries();
+                            // message size
+                            int messageSize = 0;
+                            messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId());
 
-                        Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
-                        messageBuilder.setBatchId(message.getId());
-                        if (message.getId() != -1) {
-                            if (message.isRaw()) {
-                                // for performance
-                                if (!CollectionUtils.isEmpty(message.getRawEntries())) {
+                            int dataSize = 0;
+                            for (int i = 0; i < rowEntries.size(); i++) {
+                                dataSize += com.google.protobuf.CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i));
+                            }
+                            messageSize += dataSize;
+                            messageSize += 1 * rowEntries.size();
+                            // packet size
+                            int size = 0;
+                            size += com.google.protobuf.CodedOutputStream.computeEnumSize(3,
+                                PacketType.MESSAGES.getNumber());
+                            size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
+                                    + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize)
+                                    + messageSize;
+                            // recyle bytes
+                            // ByteBuffer byteBuffer = (ByteBuffer)
+                            // ctx.getAttachment();
+                            // if (byteBuffer != null && size <=
+                            // byteBuffer.capacity()) {
+                            // byteBuffer.clear();
+                            // } else {
+                            // byteBuffer =
+                            // ByteBuffer.allocate(size).order(ByteOrder.BIG_ENDIAN);
+                            // ctx.setAttachment(byteBuffer);
+                            // }
+                            // CodedOutputStream output =
+                            // CodedOutputStream.newInstance(byteBuffer);
+                            byte[] body = new byte[size];
+                            CodedOutputStream output = CodedOutputStream.newInstance(body);
+                            output.writeEnum(3, PacketType.MESSAGES.getNumber());
+
+                            output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED);
+                            output.writeRawVarint32(messageSize);
+                            // message
+                            output.writeInt64(1, message.getId());
+                            for (int i = 0; i < rowEntries.size(); i++) {
+                                output.writeBytes(2, rowEntries.get(i));
+                            }
+                            output.checkNoSpaceLeft();
+                            NettyUtils.write(ctx.getChannel(), body, null);
+                            
+                            // output.flush();
+                            // byteBuffer.flip();
+                            // NettyUtils.write(ctx.getChannel(), byteBuffer,
+                            // null);
+                        } else {
+                            Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
+                            packetBuilder.setType(PacketType.MESSAGES);
+
+                            Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
+                            messageBuilder.setBatchId(message.getId());
+                            if (message.getId() != -1) {
+                                if (message.isRaw() && !CollectionUtils.isEmpty(message.getRawEntries())) {
                                     messageBuilder.addAllMessages(message.getRawEntries());
-                                }
-                            } else {
-                                if (!CollectionUtils.isEmpty(message.getEntries())) {
+                                } else if (!CollectionUtils.isEmpty(message.getEntries())) {
                                     for (Entry entry : message.getEntries()) {
                                         messageBuilder.addMessages(entry.toByteString());
                                     }
                                 }
                             }
+                            packetBuilder.setBody(messageBuilder.build().toByteString());
+                            NettyUtils.write(ctx.getChannel(), packetBuilder.build().toByteArray(), null);// 输出数据
                         }
-                        packetBuilder.setBody(messageBuilder.build().toByteString());
-                        NettyUtils.write(ctx.getChannel(), packetBuilder.build().toByteArray(), null);// 输出数据
                     } else {
                         NettyUtils.error(401,
                             MessageFormatter.format("destination or clientId is null", get.toString()).getMessage(),

server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsProvider.java

@@ -0,0 +1,24 @@
+package com.alibaba.otter.canal.spi;
+
+/**
+ * Use java service provider mechanism to provide {@link CanalMetricsService}.
+ * <pre>
+ * Example:
+ * {@code
+ *     ServiceLoader<CanalMetricsProvider> providers = ServiceLoader.load(CanalMetricsProvider.class);
+ *     List<CanalMetricsProvider> list = new ArrayList<CanalMetricsProvider>();
+ *     for (CanalMetricsProvider provider : providers) {
+ *         list.add(provider);
+ *     }
+ * }
+ * </pre>
+ * @author Chuanyi Li
+ */
+public interface CanalMetricsProvider {
+
+    /**
+     * @return Impl of {@link CanalMetricsService}
+     */
+    CanalMetricsService getService();
+
+}

server/src/main/java/com/alibaba/otter/canal/spi/CanalMetricsService.java

@@ -0,0 +1,42 @@
+package com.alibaba.otter.canal.spi;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+
+/**
+ * Canal server/instance metrics for export.
+ * <strong>
+ *     Designed to be created by service provider.
+ * </strong>
+ * @see CanalMetricsProvider
+ * @author Chuanyi Li
+ */
+public interface CanalMetricsService {
+
+    /**
+     * Initialization on canal server startup.
+     */
+    void initialize();
+
+    /**
+     * Clean-up at canal server stop phase.
+     */
+    void terminate();
+
+    /**
+     * @return {@code true} if the metrics service is running, otherwise {@code false}.
+     */
+    boolean isRunning();
+
+    /**
+     * Register instance level metrics for specified instance.
+     * @param instance {@link CanalInstance}
+     */
+    void register(CanalInstance instance);
+
+    /**
+     * Unregister instance level metrics for specified instance.
+     * @param instance {@link CanalInstance}
+     */
+    void unregister(CanalInstance instance);
+
+}

server/src/main/java/com/alibaba/otter/canal/spi/NopCanalMetricsService.java

@@ -0,0 +1,38 @@
+package com.alibaba.otter.canal.spi;
+
+import com.alibaba.otter.canal.instance.core.CanalInstance;
+
+/**
+ * @author Chuanyi Li
+ */
+public class NopCanalMetricsService implements CanalMetricsService {
+
+    public static final NopCanalMetricsService NOP = new NopCanalMetricsService();
+
+    private NopCanalMetricsService() {}
+
+    @Override
+    public void initialize() {
+
+    }
+
+    @Override
+    public void terminate() {
+
+    }
+
+    @Override
+    public boolean isRunning() {
+        return false;
+    }
+
+    @Override
+    public void register(CanalInstance instance) {
+
+    }
+
+    @Override
+    public void unregister(CanalInstance instance) {
+
+    }
+}

server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithEmbedded_FileModeTest.java

@@ -0,0 +1,53 @@
+package com.alibaba.otter.canal.server;
+
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+
+import com.alibaba.otter.canal.instance.manager.model.Canal;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.*;
+
+public class CanalServerWithEmbedded_FileModeTest extends BaseCanalServerWithEmbededTest {
+
+    protected Canal buildCanal() {
+        Canal canal = new Canal();
+        canal.setId(1L);
+        canal.setName(DESTINATION);
+        canal.setDesc("my standalone server test ");
+
+        CanalParameter parameter = new CanalParameter();
+
+        parameter.setMetaMode(MetaMode.LOCAL_FILE);
+        parameter.setDataDir("./conf");
+        parameter.setMetaFileFlushPeriod(1000);
+        parameter.setHaMode(HAMode.HEARTBEAT);
+        parameter.setIndexMode(IndexMode.MEMORY_META_FAILBACK);
+
+        parameter.setStorageMode(StorageMode.MEMORY);
+        parameter.setMemoryStorageBufferSize(32 * 1024);
+
+        parameter.setSourcingType(SourcingType.MYSQL);
+        parameter.setDbAddresses(Arrays.asList(new InetSocketAddress(MYSQL_ADDRESS, 3306),
+            new InetSocketAddress(MYSQL_ADDRESS, 3306)));
+        parameter.setDbUsername(USERNAME);
+        parameter.setDbPassword(PASSWORD);
+        parameter.setPositions(Arrays.asList("{\"journalName\":\"mysql-bin.000001\",\"position\":332L,\"timestamp\":\"1505998863000\"}",
+            "{\"journalName\":\"mysql-bin.000001\",\"position\":332L,\"timestamp\":\"1505998863000\"}"));
+
+        parameter.setSlaveId(1234L);
+
+        parameter.setDefaultConnectionTimeoutInSeconds(30);
+        parameter.setConnectionCharset("UTF-8");
+        parameter.setConnectionCharsetNumber((byte) 33);
+        parameter.setReceiveBufferSize(8 * 1024);
+        parameter.setSendBufferSize(8 * 1024);
+
+        parameter.setDetectingEnable(false);
+        parameter.setDetectingIntervalInSeconds(10);
+        parameter.setDetectingRetryTimes(3);
+        parameter.setDetectingSQL(DETECTING_SQL);
+
+        canal.setCanalParameter(parameter);
+        return canal;
+    }
+}

server/src/test/java/com/alibaba/otter/canal/server/ProtocolTest.java

@@ -0,0 +1,92 @@
+package com.alibaba.otter.canal.server;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+
+import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
+import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
+import com.alibaba.otter.canal.protocol.CanalEntry.Header;
+import com.alibaba.otter.canal.protocol.CanalPacket.Compression;
+import com.alibaba.otter.canal.protocol.CanalPacket.Messages;
+import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
+import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
+import com.alibaba.otter.canal.protocol.Message;
+import com.alibaba.otter.canal.protocol.exception.CanalClientException;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.CodedOutputStream;
+import com.google.protobuf.WireFormat;
+
+public class ProtocolTest {
+
+    @Test
+    public void testSimple() throws IOException {
+        Header.Builder headerBuilder = Header.newBuilder();
+        headerBuilder.setLogfileName("mysql-bin.000001");
+        headerBuilder.setLogfileOffset(1024);
+        headerBuilder.setExecuteTime(1024);
+        Entry.Builder entryBuilder = Entry.newBuilder();
+        entryBuilder.setHeader(headerBuilder.build());
+        entryBuilder.setEntryType(EntryType.ROWDATA);
+        Entry entry = entryBuilder.build();
+        Message message = new Message(3, true, Arrays.asList(entry.toByteString()));
+
+        byte[] body = buildData(message);
+        Packet packet = Packet.parseFrom(body);
+        switch (packet.getType()) {
+            case MESSAGES: {
+                if (!packet.getCompression().equals(Compression.NONE)) {
+                    throw new CanalClientException("compression is not supported in this connector");
+                }
+
+                Messages messages = Messages.parseFrom(packet.getBody());
+                Message result = new Message(messages.getBatchId());
+                for (ByteString byteString : messages.getMessagesList()) {
+                    result.addEntry(Entry.parseFrom(byteString));
+                }
+
+                System.out.println(result);
+                break;
+            }
+            default: {
+                throw new CanalClientException("unexpected packet type: " + packet.getType());
+            }
+        }
+    }
+
+    private byte[] buildData(Message message) throws IOException {
+        List<ByteString> rowEntries = message.getRawEntries();
+        // message size
+        int messageSize = 0;
+        messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId());
+
+        int dataSize = 0;
+        for (int i = 0; i < rowEntries.size(); i++) {
+            dataSize += com.google.protobuf.CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i));
+        }
+        messageSize += dataSize;
+        messageSize += 1 * rowEntries.size();
+        // packet size
+        int size = 0;
+        size += com.google.protobuf.CodedOutputStream.computeEnumSize(3, PacketType.MESSAGES.getNumber());
+        size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
+                + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize) + messageSize;
+        // TODO recyle bytes[]
+        byte[] body = new byte[size];
+        CodedOutputStream output = CodedOutputStream.newInstance(body);
+        output.writeEnum(3, PacketType.MESSAGES.getNumber());
+
+        output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED);
+        output.writeRawVarint32(messageSize);
+        // message
+        output.writeInt64(1, message.getId());
+        for (int i = 0; i < rowEntries.size(); i++) {
+            output.writeBytes(2, rowEntries.get(i));
+        }
+        output.checkNoSpaceLeft();
+
+        return body;
+    }
+}