| | |
|---|
| | | package serf |
|---|
| | | |
|---|
| | | import ( |
|---|
| | | "basic.com/valib/logger.git" |
|---|
| | | "basic.com/valib/serf.git/serf" |
|---|
| | | "encoding/json" |
|---|
| | | "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" |
|---|
| | | ) |
|---|
| | | |
|---|
| | | 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 |
|---|
| | | } |
|---|
| | | |
|---|
| | | //收到其它节点主动将注册中心的所有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] |
|---|
| | | leaveSql := "update cluster_node set isDelete=1 where node_id='" + leaveMember.Name + "'" |
|---|
| | | flag, e := executeSqlByGorm([]string{leaveSql}) |
|---|
| | | |
|---|
| | | logger.Info("EventMemberLeave,current Members:", ev.Members) |
|---|
| | | 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 { |
|---|
| | | leaveMember := ev.Members[0] |
|---|
| | | joinSql := "update cluster_node set isDelete=0 where node_id='" + leaveMember.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 + "','" + leaveMember.Name + "'," + logResult + ",'" + logErr + "')"}) |
|---|
| | | } |
|---|
| | | } |
|---|
| | | package serf
|
|---|
| | |
|
|---|
| | | import (
|
|---|
| | | "basic.com/valib/logger.git"
|
|---|
| | | "basic.com/valib/serf.git/serf"
|
|---|
| | | "encoding/json"
|
|---|
| | | "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"
|
|---|
| | | )
|
|---|
| | |
|
|---|
| | | 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
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | //收到其它节点主动将注册中心的所有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]
|
|---|
| | | leaveSql := "update cluster_node set isDelete=1 where node_id='" + leaveMember.Name + "'"
|
|---|
| | | flag, e := executeSqlByGorm([]string{leaveSql})
|
|---|
| | |
|
|---|
| | | logger.Info("EventMemberLeave,current Members:", ev.Members)
|
|---|
| | | 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 {
|
|---|
| | | leaveMember := ev.Members[0]
|
|---|
| | | joinSql := "update cluster_node set isDelete=0 where node_id='" + leaveMember.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 + "','" + leaveMember.Name + "'," + logResult + ",'" + logErr + "')"})
|
|---|
| | | }
|
|---|
| | | }
|
|---|
