package serf
|
|
import (
|
"basic.com/pubsub/protomsg.git"
|
"basic.com/valib/logger.git"
|
"basic.com/valib/serf.git/serf"
|
"encoding/json"
|
"fmt"
|
"github.com/golang/protobuf/proto"
|
"github.com/hashicorp/memberlist"
|
"github.com/satori/go.uuid"
|
"path/filepath"
|
"reflect"
|
"runtime"
|
"strconv"
|
"strings"
|
"time"
|
"vamicro/config"
|
"vamicro/system-service/bhome_msg_dev"
|
"vamicro/system-service/models"
|
)
|
|
type RpcHandle func(arg RpcParamTopic) ([]serf.NodeResponse, error)
|
|
var rpcHandlers map[string]RpcHandle
|
|
func init() {
|
rpcHandlers = make(map[string]RpcHandle)
|
}
|
|
// RegisterRpcHandles
|
func RegisterRpcHandles(fs ...RpcHandle) {
|
for _, f := range fs {
|
name := runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name() //absolute package path name eg: a.b.c.d.FuncName
|
nameEnd := filepath.Ext(name)
|
name = strings.TrimPrefix(nameEnd, ".")
|
if _, ok := rpcHandlers[name]; !ok {
|
rpcHandlers[name] = f
|
}
|
}
|
}
|
|
/*****************************UserEvent***************************************/
|
func HandleUserEventSyncSql(ev serf.UserEvent) {
|
logger.Info("receive a UserEventSyncSql event")
|
var sqlUe SqlUserEvent
|
err := json.Unmarshal(ev.Payload, &sqlUe)
|
if err != nil {
|
logger.Error("sqlUe unmarshal err:", err)
|
return
|
}
|
|
logger.Info("ev.LTime:", ev.LTime, "owner:", sqlUe.Owner, "sql:", sqlUe.Sql)
|
if sqlUe.Owner != config.Server.AnalyServerId {
|
go func() {
|
flag, e := executeSqlByGorm(sqlUe.Sql)
|
logger.Info("ev.LTime:", ev.LTime, "userEvent exec ", sqlUe.Sql, ",Result:", flag, ", err:", e)
|
logLT := strconv.Itoa(int(ev.LTime))
|
logT := time.Now().Format("2006-01-02 15:04:05")
|
logSql := strings.ReplaceAll(strings.Join(sqlUe.Sql, ";"), "'", "''")
|
logResult := "0"
|
if flag {
|
logResult = "1"
|
}
|
logErr := ""
|
if e != nil {
|
logErr = e.Error()
|
}
|
executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + sqlUe.Owner + "'," + logResult + ",'" + logErr + "')"})
|
}()
|
}
|
}
|
|
func HandleUserEventSyncDbTablePersonCache(ev serf.UserEvent) {
|
logger.Info("LTime:", ev.LTime, ",ev.Payload.len:", len(ev.Payload))
|
SyncDbTablePersonCacheChan <- ev.Payload
|
}
|
|
func HandleUserEventSyncVirtualIp(ev serf.UserEvent) {
|
logger.Info("LTime:", ev.LTime, " Recevie virtualIp change")
|
SyncVirtualIpChan <- ev.Payload
|
}
|
func HandleUserEventChangeMaster(ev serf.UserEvent) {
|
masterId := string(ev.Payload)
|
localId := Agent.LocalMember().Name
|
logger.Info("变为主节点的id,", masterId, "本节点id:", localId)
|
if masterId == localId {
|
return
|
}
|
|
var clusterDb models.Node
|
localNode, err := clusterDb.FindNodeById(localId)
|
if err == nil {
|
if localNode.DriftState == "master" {
|
logger.Info("本节点为之前的master,通知变更为slave")
|
|
// 通知主节点变更
|
chMsg := protomsg.DbChangeMessage{
|
Id: localNode.ClusterId,
|
Table: protomsg.TableChanged_T_Cluster,
|
Action: protomsg.DbAction_Insert,
|
Info: "master2slave",
|
}
|
|
bts, _ := json.Marshal(chMsg)
|
h := GetBusHandle()
|
if h == nil {
|
logger.Error("HandleEventMemberLeave bus handle is nil")
|
return
|
}
|
err := h.Publish("system-service", bts)
|
if err != nil {
|
logger.Error("HandleEventMemberLeave pub master err:", err)
|
}
|
}
|
}
|
|
clusterDb.UpdateDriftStateByNodeId("master", masterId, false)
|
}
|
|
func HandleUserEventSyncMessage(ev serf.UserEvent) {
|
logger.Info("receive a UserEventSyncMessage event")
|
var procMsg ProcMessageEvent
|
err := json.Unmarshal(ev.Payload, &procMsg)
|
if err != nil {
|
logger.Error("sqlUe unmarshal err:", err)
|
return
|
}
|
|
// 自己发送的消息不处理
|
if procMsg.Owner != config.Server.AnalyServerId {
|
// 判断是否有指定的接收目标
|
if procMsg.Target == "" || procMsg.Target == config.Server.AnalyServerId {
|
SyncProcMessageChan <- ev.Payload
|
}
|
}
|
}
|
|
// 收到其它节点主动将注册中心的所有topic通知到集群中
|
func HandleSyncRegisterInfo(ev serf.UserEvent) {
|
logger.Debug("HandleSyncRegisterInfo")
|
var si bhome_msg_dev.MsgDevRegisterInfo
|
if err := proto.Unmarshal(ev.Payload, &si); err == nil {
|
logger.Debug("HandleSyncRegisterInfo si.DevId:", string(si.DevId), " config.Server.AnalyServerId:", config.Server.AnalyServerId)
|
if string(si.DevId) != config.Server.AnalyServerId {
|
compareRPool(&si)
|
}
|
} else {
|
logger.Error("HandleSyncRegisterInfo unmarshal err:", err)
|
}
|
}
|
|
func HandleDataSystemSerfSub(ev serf.UserEvent) {
|
h := GetBusHandle()
|
if h == nil {
|
logger.Error("HandleDataSystemSerfSub bus handle is nil")
|
return
|
}
|
err := h.Publish(DataSystemSerfSubscribe, ev.Payload)
|
if err != nil {
|
logger.Error("HandleDataSystemSerfSub pub err:", err)
|
}
|
}
|
|
/*****************************Query***************************************/
|
func HandleQueryEventUpdateDBData(ev *serf.Query) {
|
logger.Info("receive QueryEventUpdateDBData, current node:", config.Server.AnalyServerId)
|
var fromP QueryTableDataParam
|
err := json.Unmarshal(ev.Payload, &fromP)
|
if err != nil {
|
logger.Error("Query tableNames unmarshal err")
|
if err := ev.Respond([]byte("request unmarshal err")); err != nil {
|
logger.Error("query.Respond err: %s\n", err)
|
return
|
}
|
|
return
|
}
|
logger.Info("Query tableNames:", fromP.Tables)
|
datas, err := DumpTables(fromP.Tables)
|
if err != nil {
|
logger.Error("queryByGorm err:", err)
|
if err := ev.Respond([]byte("queryByGorm err")); err != nil {
|
logger.Error("query.Respond err: %s\n", err)
|
return
|
}
|
return
|
}
|
bytesReturn, err := json.Marshal(datas)
|
logger.Info("results.len: ", len(bytesReturn))
|
|
var targetNode *memberlist.Node
|
nodes := Agent.Serf().Memberlist().Members()
|
if nodes != nil && len(nodes) > 0 {
|
for _, n := range nodes {
|
if n.Name == fromP.From {
|
targetNode = n
|
break
|
}
|
}
|
}
|
logger.Debug("targetNode:", targetNode.Name)
|
if targetNode != nil {
|
go func() {
|
addr := targetNode.Addr.String() + ":" + strconv.Itoa(TcpTransportPort)
|
sendErr := rawSendTcpMsg(addr, bytesReturn)
|
|
logLT := strconv.Itoa(int(ev.LTime))
|
logT := time.Now().Format("2006-01-02 15:04:05")
|
logSql := strings.ReplaceAll("QueryEventUpdateDBData from "+targetNode.Name, "'", "''")
|
logResult := "0"
|
logErr := ""
|
if sendErr == nil {
|
logResult = "1"
|
logger.Debug("sendToTcp success")
|
} else {
|
logErr = sendErr.Error()
|
logger.Debug("sendToTcp err:", sendErr)
|
}
|
|
executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + targetNode.Name + "'," + logResult + ",'" + logErr + "')"})
|
}()
|
} else {
|
logger.Debug("targetNode is nil")
|
}
|
}
|
|
// 处理其他的一些query请求
|
func HandleOtherQuery(ev *serf.Query) {
|
var reqBody RequestSerfTopicMsg
|
var ret []byte
|
if err := json.Unmarshal(ev.Payload, &reqBody); err != nil {
|
ret = []byte(err.Error())
|
} else {
|
if err, data := QueryLocalProc(reqBody); err != nil {
|
ret = []byte(err.Error())
|
} else {
|
b, e := json.Marshal(data)
|
if e != nil {
|
ret = []byte(e.Error())
|
} else {
|
ret = b
|
}
|
}
|
}
|
|
if err := ev.Respond(ret); err != nil {
|
logger.Debug("HandleOtherQuery err:", err)
|
return
|
}
|
}
|
|
func HandleQueryRpc(ev *serf.Query) {
|
var ret []byte
|
var arg RpcParamTopic
|
err := json.Unmarshal(ev.Payload, &arg)
|
if err == nil {
|
if f, ok := rpcHandlers[arg.Topic]; ok {
|
resp, e := f(arg)
|
if e == nil {
|
if data, me := json.Marshal(resp); me == nil {
|
ret = data
|
} else {
|
logger.Debug("marshal resp err:", e)
|
}
|
} else {
|
logger.Debug("call f err:", e)
|
}
|
} else {
|
logger.Debug("rpcHandlers not contains topic:", arg.Topic)
|
}
|
} else {
|
logger.Debug("unmarshal RpcParamTopic err:", err)
|
}
|
if rErr := ev.Respond(ret); rErr != nil {
|
logger.Debug("HandleQueryRpc err:", rErr)
|
}
|
}
|
|
func HandleEventMemberLeave(ev serf.MemberEvent) {
|
if ev.Members != nil && len(ev.Members) == 1 {
|
leaveMember := ev.Members[0]
|
logger.Info("Event Member Leave, Members:", ev.Members)
|
|
leaveSql := "update cluster_node set isDelete=1 where node_id='" + leaveMember.Name + "'"
|
flag, e := executeSqlByGorm([]string{leaveSql})
|
|
// 判断离开是否是主节点离开, 更换主节点
|
var clusterDb models.Node
|
leaveNode, err := clusterDb.FindNodeById(leaveMember.Name)
|
if err == nil {
|
logger.Info("查询离开节点的信息,", leaveNode)
|
if leaveNode.DriftState == "master" {
|
firstNode, _ := clusterDb.FindFirstNode()
|
logger.Info("离开的节点为主节点, 查询加入最早的节点, ", firstNode)
|
logger.Info("本节点信息:", Agent.LocalMember().Name)
|
if firstNode.NodeId == Agent.LocalMember().Name {
|
logger.Info("更新本节点为主节点")
|
clusterDb.UpdateDriftStateByNodeId("master", firstNode.NodeId, true)
|
|
// 通知主节点变更
|
chMsg := protomsg.DbChangeMessage{
|
Id: firstNode.ClusterId,
|
Table: protomsg.TableChanged_T_Cluster,
|
Action: protomsg.DbAction_Insert,
|
Info: "slave2master",
|
}
|
|
bts, _ := json.Marshal(chMsg)
|
h := GetBusHandle()
|
if h == nil {
|
logger.Error("HandleEventMemberLeave bus handle is nil")
|
return
|
}
|
err := h.Publish("system-service", bts)
|
if err != nil {
|
logger.Error("HandleEventMemberLeave pub master err:", err)
|
}
|
}
|
}
|
}
|
|
logLT := ""
|
logT := time.Now().Format("2006-01-02 15:04:05")
|
logSql := strings.ReplaceAll(leaveSql, "'", "''")
|
logResult := "0"
|
if flag {
|
logResult = "1"
|
}
|
logErr := ""
|
if e != nil {
|
logErr = e.Error()
|
}
|
executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + leaveMember.Name + "'," + logResult + ",'" + logErr + "')"})
|
}
|
}
|
|
func HandleEventMemberJoin(ev serf.MemberEvent) {
|
if ev.Members != nil && len(ev.Members) == 1 {
|
joinMember := ev.Members[0]
|
|
timeUnix := time.Now().Unix()
|
fmtTimeStr := time.Unix(timeUnix, 0).Format("2006-01-02 15:04:05")
|
|
// 更新在线状态
|
updateOnlineSql := fmt.Sprintf("update cluster_node set online=1 where node_id='%s'", joinMember.Name)
|
|
// 如果是新加入的更新创建时间
|
joinSql := fmt.Sprintf("update cluster_node set isDelete=0,drift_state='slave',node_ip='%s',create_time='%s' where node_id='%s' and isDelete=1", joinMember.Addr.String()+":"+strconv.Itoa(int(joinMember.Port)), fmtTimeStr, joinMember.Name)
|
flag, e := executeSqlByGorm([]string{updateOnlineSql, joinSql})
|
|
logger.Info("EventMemberJoin,current Members:", ev.Members)
|
logLT := ""
|
logT := time.Now().Format("2006-01-02 15:04:05")
|
logSql := strings.ReplaceAll(joinSql, "'", "''")
|
logResult := "0"
|
if flag {
|
logResult = "1"
|
}
|
logErr := ""
|
if e != nil {
|
logErr = e.Error()
|
}
|
executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + joinMember.Name + "'," + logResult + ",'" + logErr + "')"})
|
}
|
}
|
|
func HandleEventMemberFail(ev serf.MemberEvent) {
|
if ev.Members != nil && len(ev.Members) == 1 {
|
joinMember := ev.Members[0]
|
|
//joinSql := "update cluster_node set isDelete=0 where node_id='" + leaveMember.Name + "'"
|
joinSql := fmt.Sprintf("update cluster_node set online=0 where node_id='%s'", joinMember.Name)
|
flag, e := executeSqlByGorm([]string{joinSql})
|
|
logger.Info("EventMemberJoin,current Members:", ev.Members)
|
logLT := ""
|
logT := time.Now().Format("2006-01-02 15:04:05")
|
logSql := strings.ReplaceAll(joinSql, "'", "''")
|
logResult := "0"
|
if flag {
|
logResult = "1"
|
}
|
logErr := ""
|
if e != nil {
|
logErr = e.Error()
|
}
|
executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + joinMember.Name + "'," + logResult + ",'" + logErr + "')"})
|
}
|
}
|
|
func HandleUpdateMemberStatus() {
|
if Agent == nil {
|
return
|
}
|
|
for _, member := range Agent.Serf().Members() {
|
alive := 0
|
if member.Status == serf.StatusAlive {
|
alive = 1
|
}
|
|
sql := fmt.Sprintf("update cluster_node set online=%d where node_id='%s'", alive, member.Name)
|
_, err := executeSqlByGorm([]string{sql})
|
if err != nil {
|
logger.Error(err)
|
}
|
}
|
}
|
