package serf
|
|
import (
|
"context"
|
"encoding/json"
|
"fmt"
|
"os"
|
"os/signal"
|
"regexp"
|
"strings"
|
"syscall"
|
"time"
|
|
"apsClient/pkg/logx"
|
|
"basic.com/pubsub/protomsg.git"
|
"basic.com/valib/bhomeclient.git"
|
"basic.com/valib/bhomedbapi.git"
|
|
"github.com/gogo/protobuf/proto"
|
"github.com/jinzhu/gorm"
|
"github.com/satori/go.uuid"
|
"github.com/mitchellh/mapstructure"
|
"github.com/muesli/cache2go"
|
)
|
|
var (
|
agent = SyncServer{}
|
dependProcs = []string{
|
bhomeclient.Proc_System_Service,
|
}
|
|
sqlMsgSeqCache = cache2go.Cache("syncSqlMsg")
|
)
|
|
const (
|
serfSyncTopic = "sync-proc-message-to-serf"
|
|
EventCreateCluster = 0
|
EventJoinCluster = 1
|
EventLeaveCluster = 2
|
EventMaster2Slave = 3
|
EventSlave2Master = 4
|
)
|
|
type ProcMessageEvent struct {
|
Owner string `json:"owner"` // 发送者
|
Target string `json:"target"` // 指定接收者
|
Proc string `json:"procName"` // 进程名
|
Topic string `json:"topic"` // 主题
|
Payload []byte `json:"payload"` // 消息体,自行解析
|
}
|
|
type SqlMsg struct {
|
Id string
|
Sql string
|
Version string
|
}
|
|
type SyncServer struct {
|
ProcName string // 进程名称
|
ServerId string // 本机id
|
ClusterStatus string // 集群状态 master/slave 为空表示未加入集群
|
syncSqlTopic string // 同步sql消息的主题
|
queryTableTopic string // 加入集群后请求集群数据的主题
|
syncTables []string // 需要同步的表
|
sqlDB *gorm.DB // 数据库
|
bhClient *bhomeclient.MicroNode
|
clusterEventFn func(int)
|
}
|
|
func InitAgent(procName string, syncTables []string, db *gorm.DB) *SyncServer {
|
agent.ProcName = procName
|
agent.ServerId = Vasystem.ServerID
|
agent.sqlDB = db
|
agent.syncTables = syncTables
|
agent.syncSqlTopic = procName + "/serf/sync/sql"
|
agent.queryTableTopic = procName + "/serf/query/sqls"
|
|
// 设置日志回调
|
db.SetLogger(&agent)
|
|
// 先关闭日志
|
//db.LogMode(false)
|
|
return &agent
|
}
|
|
func (ss *SyncServer) RegisterClusterEvent(fn func(int)) {
|
ss.clusterEventFn = fn
|
}
|
|
func (ss *SyncServer) Serve(initChan chan bool) {
|
proc := &bhomeclient.ProcInfo{
|
Name: ss.ProcName, //进程名称
|
ID: ss.ProcName, //进程id
|
Info: "", //进程的描述信息，用于区分同一进程名称下多个进程
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
var reg = &bhomeclient.RegisterInfo{
|
Proc: *proc,
|
Channel: nil,
|
PubTopic: []string{},
|
SubTopic: []string{bhomeclient.Proc_System_Service, ss.syncSqlTopic, ss.queryTableTopic},
|
SubNetTopic: []string{},
|
}
|
|
q := make(chan os.Signal, 1)
|
signal.Notify(q, os.Interrupt, os.Kill, syscall.SIGTERM)
|
|
client, err := bhomeclient.NewMicroNode(ctx, q, ss.ServerId, reg, nil)
|
if err != nil {
|
initChan <- false
|
return
|
}
|
|
bhomedbapi.InitGetNetNode(client.GetLocalNetNodeByTopic)
|
bhomedbapi.InitDoReq(client.RequestOnly)
|
//bhomedbapi.InitLog(logger.Debug)
|
|
// 需要等待system-service进程成功启动后，才能获取集群状态(或者保证程序启动时获取到正确的状态)
|
tryTimes := 0
|
loop:
|
for {
|
select {
|
case <-q:
|
initChan <- false
|
return
|
default:
|
if tryTimes < 15 {
|
clients, err := client.GetRegisteredClient()
|
if err == nil && len(clients) > 0 {
|
var existingProcs []string
|
for _, c := range clients {
|
if c.Online {
|
existingProcs = append(existingProcs, string(c.Proc.ProcId))
|
}
|
}
|
if diff := arrayContains(existingProcs, dependProcs); diff == "" {
|
break loop
|
} else {
|
logx.Errorf("Proc: %s is not running!", diff)
|
time.Sleep(time.Second * 1)
|
}
|
} else {
|
tryTimes++
|
time.Sleep(time.Second * 5)
|
}
|
} else {
|
logx.Errorf("tried 15 times, client.GetRegisteredClient failed")
|
initChan <- false
|
return
|
}
|
}
|
}
|
|
go client.StartServer(nil)
|
|
ss.bhClient = client
|
|
go ss.subBusMessage(ctx)
|
|
go ss.handleDbLoggerPrint()
|
|
// 启动后查询一次集群状态
|
ss.QueryClusterStat()
|
|
//if ss.ClusterStatus != "" {
|
//ss.sqlDB.LogMode(true)
|
//}
|
|
initChan <- true
|
<-q
|
|
client.DeRegister()
|
cancel()
|
client.Free()
|
|
os.Exit(0)
|
}
|
|
func (ss *SyncServer) pubSyncSqlMessage(sql string, targetId string) error {
|
sqlMsg := SqlMsg{
|
Id: uuid.NewV4().String(),
|
Sql: sql,
|
}
|
|
bMsg, _ := json.Marshal(sqlMsg)
|
|
var msg = ProcMessageEvent{
|
Owner: ss.ServerId,
|
Target: targetId,
|
Proc: ss.ProcName,
|
Topic: ss.syncSqlTopic,
|
Payload: bMsg,
|
}
|
|
b, err := json.Marshal(msg)
|
if err != nil {
|
return err
|
}
|
|
return ss.bhClient.Publish(serfSyncTopic, b)
|
}
|
|
// 请求同步表的全量数据, 发送自己的id
|
func (ss *SyncServer) pubSyncTableMessage() error {
|
var msg = ProcMessageEvent{
|
Owner: ss.ServerId,
|
Proc: ss.ProcName,
|
Topic: ss.queryTableTopic,
|
Payload: []byte(ss.ServerId),
|
}
|
|
b, err := json.Marshal(msg)
|
if err != nil {
|
return err
|
}
|
|
logx.Debugf("加入集群, 请求同步全量数据,id:%s", ss.ServerId)
|
return ss.bhClient.Publish(serfSyncTopic, b)
|
}
|
|
func (ss *SyncServer) subBusMessage(ctx context.Context) {
|
//fmt.Println("sub bus msg")
|
|
for {
|
select {
|
case <-ctx.Done():
|
fmt.Println("sub bus msg exit")
|
return
|
case busMsg := <-ss.bhClient.SubCh:
|
if string(busMsg.Topic) == ss.syncSqlTopic {
|
ss.handleClusterMessage(busMsg.Data)
|
}
|
|
// 处理同步全量数据的请求
|
if string(busMsg.Topic) == ss.queryTableTopic {
|
if ss.ClusterStatus == "master" {
|
logx.Debugf("接收到同步全量数据请求.")
|
ss.handleSyncTableMessage(busMsg.Data)
|
}
|
}
|
|
// system-service发送的消息
|
if string(busMsg.Topic) == bhomeclient.Proc_System_Service {
|
var clusterMsg = &protomsg.DbChangeMessage{}
|
|
if err := proto.Unmarshal(busMsg.Data, clusterMsg); err != nil {
|
if err = json.Unmarshal(busMsg.Data, clusterMsg); err != nil {
|
fmt.Println("proto.Unmarshal ", err.Error())
|
continue
|
}
|
}
|
|
if clusterMsg.Table == protomsg.TableChanged_T_Cluster {
|
switch clusterMsg.Info {
|
case "create":
|
// 创建集群, 开启日志跟踪, 设置角色master
|
ss.clusterEventFn(EventCreateCluster)
|
ss.ClusterStatus = "master"
|
//ss.sqlDB.LogMode(true)
|
|
case "join":
|
// 加入集群, 开启日志跟踪, 设置角色slave
|
ss.clusterEventFn(EventJoinCluster)
|
ss.onJoinCluster()
|
ss.ClusterStatus = "slave"
|
//ss.sqlDB.LogMode(true)
|
|
case "leave":
|
// 退出集群, 开启日志跟踪, 设置角色slave
|
ss.clusterEventFn(EventLeaveCluster)
|
ss.ClusterStatus = ""
|
//ss.sqlDB.LogMode(true)
|
case "slave2master":
|
ss.clusterEventFn(EventSlave2Master)
|
ss.ClusterStatus = "master"
|
//ss.sqlDB.LogMode(true)
|
case "master2slave":
|
ss.clusterEventFn(EventMaster2Slave)
|
ss.ClusterStatus = "slave"
|
//ss.sqlDB.LogMode(true)
|
}
|
}
|
}
|
}
|
}
|
}
|
|
// 加入集群, 清空本地表, 同步集群内数据
|
func (ss *SyncServer) onJoinCluster() {
|
var err error
|
|
db := ss.sqlDB
|
|
tx := db.Begin()
|
defer func() {
|
if err != nil && tx != nil {
|
tx.Rollback()
|
}
|
}()
|
|
tx.Exec("PRAGMA foreign_keys=OFF")
|
//1.删除本地的同步库数据
|
for _, t := range ss.syncTables {
|
delSql := "delete from " + t + ""
|
|
err = tx.Exec(delSql).Error
|
if err != nil {
|
logx.Errorf("删除本地的同步库数据失败, %s", err.Error())
|
}
|
}
|
|
//4.开启reference
|
tx.Exec("PRAGMA foreign_keys=ON")
|
tx.Commit()
|
|
// 拉取集群内的同步库数据到本地数据库表中
|
ss.pubSyncTableMessage()
|
}
|
|
func (ss *SyncServer) onLeaveCluster() {
|
|
}
|
|
func (ss *SyncServer) onCreateCluster() {
|
|
}
|
|
// 查询集群状态, 返回 master, slave, leave
|
func (ss *SyncServer) QueryClusterStat() *bhomeclient.Reply {
|
clusterStatTopic := "/data/api-v/cluster/status"
|
req := bhomeclient.Request{
|
Path: clusterStatTopic,
|
Method: "POST",
|
}
|
|
reply, err := ss.bhClient.RequestTopic(ss.ServerId, req, 3000)
|
if err != nil {
|
fmt.Println("RequestTopic error", err.Error())
|
|
return reply
|
}
|
|
ss.ClusterStatus = reply.Msg
|
|
logx.Debugf("当前集群状态: %s", ss.ClusterStatus)
|
|
return reply
|
}
|
|
func (ss *SyncServer) handleDbLoggerPrint() {
|
sqlBuf := make([]string, 0)
|
ticker := time.NewTicker(3 * time.Second)
|
sendSize := 0 //serf MaxUserEventSize is 9*1024
|
for {
|
select {
|
case <-ticker.C:
|
if len(sqlBuf) > 0 {
|
syncSql := strings.Join(sqlBuf, "")
|
|
//fmt.Println("同步sql语句:", syncSql)
|
ss.pubSyncSqlMessage(syncSql, "")
|
|
sqlBuf = append([]string{})
|
sendSize = 0
|
}
|
case sql := <-syncSqlChan:
|
if sendSize+len(sql) > (9*1024 - 1024) {
|
if len(sqlBuf) > 0 {
|
syncSql := strings.Join(sqlBuf, "")
|
//fmt.Println("同步sql语句:", syncSql)
|
|
ss.pubSyncSqlMessage(syncSql, "")
|
|
sqlBuf = append([]string{})
|
}
|
|
s := strings.TrimRight(sql, ";")
|
sqlBuf = append(sqlBuf, s+";")
|
sendSize = len(sql)
|
} else {
|
s := strings.TrimRight(sql, ";")
|
sqlBuf = append(sqlBuf, s+";")
|
|
sendSize = sendSize + len(sql)
|
}
|
}
|
}
|
}
|
|
func (ss *SyncServer) handleClusterMessage(clusterMsgData []byte) {
|
var msg SqlMsg
|
err := json.Unmarshal(clusterMsgData,&msg)
|
if err != nil {
|
logx.Errorf(" Unmarshal cluster message error, %s",err.Error())
|
return
|
}
|
|
// 判断消息是否曾经接收过
|
if sqlMsgSeqCache.Exists(msg.Id) {
|
logx.Infof("clusterMessage:接收到重复消息, %s", msg.Sql)
|
return
|
}
|
|
// 记录消息id, 半小时过期
|
sqlMsgSeqCache.Add(msg.Id, 30*time.Minute, true)
|
|
logx.Infof("clusterMessage:%s", msg.Sql)
|
sql := msg.Sql
|
|
if len(sql) <= 0 {
|
return
|
}
|
|
db := ss.sqlDB
|
if db != nil {
|
db.LogMode(false)
|
defer db.LogMode(true)
|
|
var err error
|
tx := db.Begin()
|
defer func() {
|
if err != nil && tx != nil {
|
tx.Rollback()
|
}
|
}()
|
result := tx.Exec(sql)
|
err = result.Error
|
if err != nil {
|
fmt.Println("ExecuteSqlByGorm err:", err, ",sql:", sql)
|
}
|
if result.RowsAffected == 0 {
|
fmt.Println("ExecuteSqlByGorm RowsAffected == 0", ",sql:", sql)
|
}
|
tx.Commit()
|
}
|
}
|
|
// serf 同步数据的限制为92160 byte
|
func (ss *SyncServer) handleSyncTableMessage(msg []byte) error {
|
sizeLimit := 61440
|
targetId := string(msg)
|
|
//fmt.Println("同步全量数据给节点:", targetId)
|
sqls, err := DumpTables(ss.sqlDB, ss.syncTables)
|
if err != nil {
|
logx.Errorf("DumpTables error: %s", err.Error())
|
return err
|
}
|
|
logx.Infof("DumpTables sql:%v", sqls)
|
syncSql := strings.Join(sqls, ";")
|
if len(syncSql) < sizeLimit {
|
err = ss.pubSyncSqlMessage(syncSql, targetId)
|
} else {
|
shard := ""
|
for _, sql := range sqls {
|
if len(shard)+len(sql) > sizeLimit {
|
err = ss.pubSyncSqlMessage(shard, targetId)
|
shard = ""
|
}
|
|
shard = fmt.Sprintf("%s%s;", shard, sql)
|
}
|
|
if len(shard) > 0 {
|
err = ss.pubSyncSqlMessage(shard, targetId)
|
}
|
}
|
|
return err
|
}
|
|
func (ss *SyncServer) Print(values ...interface{}) {
|
var (
|
level = values[0]
|
)
|
|
//fmt.Println("dblogger", values)
|
|
if level == "sql" {
|
msgArr := gorm.LogFormatter(values...)
|
sql := msgArr[3].(string)
|
logx.Infof("sql: %v", sql)
|
sql = strings.TrimPrefix(sql, " ")
|
if !strings.HasPrefix(sql, "SELECT") && !strings.HasPrefix(sql, "select") && !strings.Contains(sql, "PRAGMA") && !strings.Contains(sql, "pragma") {
|
affected := values[5].(int64)
|
if affected > 0 { //执行成功
|
//判断操作的是哪张表
|
whereIdx := strings.Index(sql, "WHERE")
|
sqlWithTable := sql
|
if whereIdx > -1 {
|
sqlWithTable = sql[:whereIdx]
|
}
|
|
//fmt.Println("判断是哪张表 sqlWithTable:", sqlWithTable)
|
|
insertReg := regexp.MustCompile(`^\s*(?i:insert)\s`) //insert
|
updateReg := regexp.MustCompile(`^\s*(?i:update)\s`) //update
|
delReg := regexp.MustCompile(`^\s*(?i:delete)\s`) //delete
|
|
if insertReg.MatchString(sqlWithTable) {
|
//fmt.Println("插入操作")
|
for _, t := range agent.syncTables {
|
reg := regexp.MustCompile(`\s+\"?(?i:` + t + `)\"?\s+`)
|
if reg.MatchString(sqlWithTable) {
|
//fmt.Println("属于同步表:", t)
|
// 判断是在集群内, 同步消息, 判断两种角色, 为避免其他出现状态
|
if ss.ClusterStatus == "master" || ss.ClusterStatus == "slave" {
|
syncSqlChan <- sql
|
}
|
}
|
}
|
} else if updateReg.MatchString(sqlWithTable) || delReg.MatchString(sqlWithTable) {
|
//fmt.Println("删除或者更新")
|
for _, t := range agent.syncTables {
|
reg := regexp.MustCompile(`\s+\"?(?i:` + t + `)\"?\s+`)
|
if reg.MatchString(sqlWithTable) {
|
//fmt.Println("属于同步表:", t)
|
if ss.ClusterStatus == "master" || ss.ClusterStatus == "slave" {
|
syncSqlChan <- sql
|
}
|
}
|
}
|
}
|
}
|
}
|
} else {
|
fmt.Println("dbLogger level!=sql")
|
}
|
}
|
|
func arrayContains(list []string, arr []string) string {
|
if arr == nil || list == nil {
|
return ""
|
}
|
|
for _, s := range arr {
|
isExist := false
|
for _, t := range list {
|
if s == t {
|
isExist = true
|
break
|
}
|
}
|
|
if !isExist {
|
return s
|
}
|
}
|
|
return ""
|
}
|
|
type NodeInfo struct {
|
NodeID string `json:"node_id,omitempty"`
|
NodeIp string `json:"node_ip,omitempty"`
|
NodeName string `json:"node_name,omitempty"`
|
ClusterID string `json:"cluster_id"`
|
CreateTime string `json:"create_time"`
|
DeviceType string `json:"device_type"`
|
DriftState string `json:"drift_state"`
|
Online bool `json:"online"`
|
}
|
|
func QueryClusterStatusAndNodeQuantity() (string, int) {
|
reply := agent.QueryClusterStat()
|
if reply == nil {
|
return "", 0
|
}
|
var nodes []NodeInfo
|
err := mapstructure.Decode(reply.Data, &nodes)
|
if err != nil {
|
logx.Errorf("mapstructure.Decode QueryClusterStat data err:%v", err)
|
return reply.Msg, 0
|
}
|
return reply.Msg, len(nodes)
|
}
|
