feat: pika cdc for incremental synchronization

.gitignore

@@ -74,3 +74,4 @@ pkg
 !codis/cmd/fe/assets/**
 
 tests/tmp
+tools/pika_cdc/pika/proto

src/pika_inner_message.proto

@@ -1,6 +1,8 @@
 syntax = "proto2";
 package InnerMessage;
 
+option go_package = "./proto/inner";
+
 enum Type {
   kMetaSync        = 1;
   kTrySync         = 2;

src/rsync_service.proto

@@ -1,6 +1,8 @@
 syntax = "proto2";
 package RsyncService;
 
+option go_package = "./proto/rsync";
+
 enum Type {
     kRsyncMeta = 1;
     kRsyncFile = 2;

tools/pika_cdc/Makefile

@@ -0,0 +1,26 @@
+PROTO_OUT = ./pika
+PROTO_DIR = ../../src/
+
+GO_FILES = $(shell find . -name '*.go')
+
+PROTO_FILES := $(wildcard $(PROTO_DIR)/*.proto)
+
+OUTPUT_BIN = pika_cdc
+
+PROTOC = protoc
+GO_BUILD = go build
+GO_CLEAN = go clean
+
+.PHONY: all proto build clean
+
+all: proto build
+
+proto: $(PROTO_FILES)
+	$(PROTOC) --proto_path=$(PROTO_DIR) --go_out=$(PROTO_OUT) $^
+
+build: $(GO_FILES)
+	$(GO_BUILD) -o $(OUTPUT_BIN)
+
+clean:
+	$(GO_CLEAN)
+	rm -f $(OUTPUT_BIN)

tools/pika_cdc/README.md

@@ -0,0 +1,24 @@
+# Pika cdc
+**A tool for incremental synchronization of pika command**
+
+By imitating a pika slave
+
+# Build
+**Make sure the system has protoc installed** 
+```bash
+brew install protobuf
+go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
+go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
+```
+
+## Build pika cdc
+```bash
+make
+```
+
+## Todo:
+
+Consumer side:
+- [x] **redis** 
+- [x] **kafka** Create a topic of the same name for each pika's DB
+- [ ] **bifrost** 

tools/pika_cdc/conf/cdc.yml

@@ -0,0 +1,20 @@
+# pika_server
+pika_server : 127.0.0.1:11221
+# For data from one DB of one pika, a separate MQ topic is created,
+# and the name of the topic is the dbname of the pika
+kafka_servers:
+  - 127.0.0.1:9092
+redis_servers:
+  - 127.0.0.1:6379
+pulsar_servers:
+  - 127.0.0.1:6650
+# retry times while send message failed
+retries : 0
+# retry interval while send message failed(ms)
+retry_interval: 10
+parallel_thread_size: 1
+# the size of the cached channel in pika cdc
+buffer_msg_numbers: 10
+
+
+

tools/pika_cdc/conf/conf.go

@@ -0,0 +1,50 @@
+package conf
+
+import (
+	"fmt"
+	"github.com/sirupsen/logrus"
+	"gopkg.in/yaml.v3"
+	"io/ioutil"
+	"log"
+	"os"
+	"path"
+	"path/filepath"
+	"runtime"
+)
+
+type PikaCdcConfig struct {
+	PikaServer         string   `yaml:"pika_server"`
+	KafkaServers       []string `yaml:"kafka_servers"`
+	RedisServers       []string `yaml:"redis_servers"`
+	PulsarServers      []string `yaml:"pulsar_servers"`
+	Retries            int      `yaml:"retries"`
+	RetryInterval      int      `yaml:"retry_interval"`
+	ParallelThreadSize int      `yaml:"parallel_thread_size"`
+	BufferMsgNumbers   int      `yaml:"buffer_msg_numbers"`
+}
+
+var ConfigInstance = PikaCdcConfig{}
+
+func init() {
+	_, filename, _, _ := runtime.Caller(0)
+	filename = filepath.Join(filepath.Dir(filename), "cdc.yml")
+	file, err := ioutil.ReadFile(filename)
+	if err != nil {
+		log.Fatal("fail to read file:", err)
+	}
+
+	err = yaml.Unmarshal(file, &ConfigInstance)
+	if err != nil {
+		log.Fatal("fail to yaml unmarshal:", err)
+	}
+
+	logrus.SetFormatter(&logrus.TextFormatter{
+		FullTimestamp: true,
+		CallerPrettyfier: func(f *runtime.Frame) (string, string) {
+			return "", fmt.Sprintf("%s:%d", path.Base(f.File), f.Line)
+		},
+	})
+
+	logrus.SetReportCaller(true)
+	logrus.SetOutput(os.Stdout)
+}

tools/pika_cdc/consumer/consumer.go

@@ -0,0 +1,32 @@
+package consumer
+
+import (
+	"pika_cdc/conf"
+)
+
+type Consumer interface {
+	SendCmdMessage(dbName string, msg []byte) error
+	Name() string
+	Close() error
+	Run()
+	Stop()
+}
+
+type Factory struct{}
+
+func GenerateConsumers(config conf.PikaCdcConfig, msgChanns map[string]chan []byte) ([]Consumer, error) {
+	var consumers []Consumer
+
+	// kafka
+	for _, k := range config.KafkaServers {
+		kafka, _ := NewKafka(k, config.Retries, msgChanns)
+		consumers = append(consumers, kafka)
+	}
+
+	// redis
+	for _, r := range config.RedisServers {
+		newRedis, _ := NewRedis(r, msgChanns)
+		consumers = append(consumers, newRedis)
+	}
+	return consumers, nil
+}

tools/pika_cdc/consumer/kafka.go

@@ -0,0 +1,91 @@
+package consumer
+
+import (
+	"context"
+	"github.com/segmentio/kafka-go"
+	"sync"
+)
+
+type Kafka struct {
+	servers    string
+	topics     []string
+	retries    int
+	kafkaConns map[string]*kafka.Conn
+	wg         sync.WaitGroup
+	msgChanns  map[string]chan []byte
+	stopChan   chan bool
+	once       sync.Once
+	protocol   KafkaProtocol
+}
+
+func (k *Kafka) SendCmdMessage(dbName string, msg []byte) error {
+	k.kafkaConns[dbName].Write(k.protocol.ToConsumer(msg))
+	return nil
+}
+
+func (k *Kafka) Name() string {
+	return "Kafka"
+}
+
+func NewKafka(server string, retries int, msgChanns map[string]chan []byte) (*Kafka, error) {
+	k := &Kafka{}
+	k.protocol = KafkaProtocol{}
+	k.kafkaConns = make(map[string]*kafka.Conn)
+	k.msgChanns = make(map[string]chan []byte)
+	for dbname, chann := range msgChanns {
+		conn, err := kafka.DialLeader(context.Background(), "tcp", server, dbname, 0)
+		if err != nil {
+			return k, err
+		} else {
+			k.kafkaConns[dbname] = conn
+		}
+		k.msgChanns[dbname] = chann
+	}
+	k.stopChan = make(chan bool)
+	k.retries = retries
+	return k, nil
+}
+
+func (k *Kafka) close() error {
+	//k.stopChan <- true
+	//close(k.stopChan)
+	//close(*k.messageChan)
+	//for _, conn := range k.kafkaConns {
+	//	err := conn.Close()
+	//	if err != nil {
+	//		logrus.Warn(err)
+	//		return err
+	//	}
+	//}
+	return nil
+}
+func (k *Kafka) Close() error {
+	//var err error
+	//err = nil
+	//k.once.Do(func() {
+	//	err = k.close()
+	//})
+	//return err
+	return nil
+}
+func (k *Kafka) Run() {
+	var wg sync.WaitGroup
+	for dbName, chann := range k.msgChanns {
+		wg.Add(1)
+		go func(dbName string, ch chan []byte) {
+			defer wg.Done()
+			for {
+				select {
+				case msg := <-ch:
+					k.SendCmdMessage(dbName, msg)
+				case <-k.stopChan:
+					return
+				}
+			}
+		}(dbName, chann)
+	}
+	wg.Wait()
+}
+func (k *Kafka) Stop() {
+	k.stopChan <- true
+}

tools/pika_cdc/consumer/protocol.go

@@ -0,0 +1,27 @@
+package consumer
+
+import (
+	"fmt"
+	"strconv"
+)
+
+type Protocol interface {
+	ToConsumer(msg []byte) []byte
+}
+type RedisProtocol struct{}
+
+func (rp RedisProtocol) ToConsumer(msg []byte) []byte {
+	return msg
+}
+func (rp RedisProtocol) Select(dbName string) []byte {
+	db, _ := strconv.Atoi(dbName[len(dbName)-1:])
+	dbStr := strconv.Itoa(db)
+	msg := fmt.Sprintf("*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", len(dbStr), dbStr)
+	return []byte(msg)
+}
+
+type KafkaProtocol struct{}
+
+func (kp KafkaProtocol) ToConsumer(msg []byte) []byte {
+	return msg
+}

tools/pika_cdc/consumer/redis.go

@@ -0,0 +1,80 @@
+package consumer
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+)
+
+type Redis struct {
+	redisProtocol RedisProtocol
+	conns         map[string]net.Conn
+	msgChanns     map[string]chan []byte
+	stopChan      chan bool
+}
+
+func NewRedis(addr string, msgChanns map[string]chan []byte) (*Redis, error) {
+	r := &Redis{redisProtocol: RedisProtocol{}, conns: make(map[string]net.Conn), msgChanns: msgChanns, stopChan: make(chan bool)}
+	var err error
+	for dbName, _ := range msgChanns {
+		r.conns[dbName], err = net.Dial("tcp", addr)
+		if err != nil {
+			return nil, fmt.Errorf("failed to connect to Redis server: %v", err)
+		}
+		selectCmdBytes := r.redisProtocol.Select(dbName)
+		r.conns[dbName].Write(selectCmdBytes)
+	}
+	return r, nil
+}
+
+func (r *Redis) SendCmdMessage(dbName string, msg []byte) error {
+	_, err := r.sendRedisData(dbName, msg)
+	return err
+}
+
+func (r *Redis) Name() string {
+	return string("Redis")
+}
+func (r *Redis) Close() error {
+	for _, conn := range r.conns {
+		conn.Close()
+	}
+	return nil
+}
+
+func (r *Redis) sendRedisData(dbName string, data []byte) (string, error) {
+	r.conns[dbName].SetDeadline(time.Now().Add(5 * time.Second))
+	_, err := r.conns[dbName].Write(data)
+	if err != nil {
+		return "", fmt.Errorf("failed to send data to Redis server: %v", err)
+	}
+	reader := bufio.NewReader(r.conns[dbName])
+	response, err := reader.ReadString('\n')
+	if err != nil {
+		return "", fmt.Errorf("failed to read response from Redis server: %v", err)
+	}
+	return response, nil
+}
+func (r *Redis) Run() {
+	var wg sync.WaitGroup
+	for dbName, chann := range r.msgChanns {
+		wg.Add(1)
+		go func(dbName string, ch chan []byte) {
+			defer wg.Done()
+			for {
+				select {
+				case msg := <-ch:
+					r.sendRedisData(dbName, msg)
+				case <-r.stopChan:
+					return
+				}
+			}
+		}(dbName, chann)
+	}
+	wg.Wait()
+}
+func (r *Redis) Stop() {
+	r.stopChan <- true
+}