Merge branch 'master' of http://192.168.1.14:10010/r/syncdb

liuxiaolong

2019-09-05 5254b8f9e86010f517432f94fa81c6c3c33e5a47

 agent.go

@@ -25,6 +25,8 @@
   "io/ioutil"
   "net"
   "os"
   "strconv"
   "sync"

   //"os"
   "strings"
@@ -39,6 +41,8 @@
const (
   QueryEventGetDB        = "GetDatabase"
   QueryEventUpdateDBData = "UpdateDBData"
   UserEventSyncSql      = "SyncSql"
   UserEventSyncDbTablePersonCache      = "SyncCache"
)

// Agent warps the serf agent
@@ -109,6 +113,8 @@
   go a.BroadcastMemberlist(BroadcastInterval * time.Second)
}

var SyncDbTablePersonCacheChan = make(chan []byte,0)

// HandleEvent Handles serf.EventMemberJoin events,
// which will wait for members to join until the number of group members is equal to "groupExpect"
// when the startup mode is "ModeCluster",
@@ -117,15 +123,22 @@

   switch ev := event.(type) {
   case serf.UserEvent:
      fmt.Println(string(ev.Payload))
      var tmpstringslice []string
      tmpstringslice = append(tmpstringslice, string(ev.Payload))
      fmt.Println(tmpstringslice)
      results, err := ExecuteWriteSql(tmpstringslice)

      for _, result := range results {
         fmt.Println(result, "results err: ", err)
      if ev.Name == UserEventSyncSql {
         var sqlUe SqlUserEvent
         err := json.Unmarshal(ev.Payload, &sqlUe)
         if err !=nil {
            fmt.Println("sqlUe unmarshal err:",err)
            return
         }
         if sqlUe.Owner != a.conf.NodeName {
            //results, err := ExecuteWriteSql(sqlArr)
            flag, _ := ExecuteSqlByGorm(sqlUe.Sql)
            fmt.Println("userEvent exec ",sqlUe.Sql,",Result:",flag)
         }
      } else if ev.Name == UserEventSyncDbTablePersonCache {
         SyncDbTablePersonCacheChan <- ev.Payload
      }


   case *serf.Query:

@@ -178,13 +191,14 @@
            fmt.Println("Query tableNames unmarshal err")
            return
         }
         fmt.Println("Query tableNames:",tableNames)
         datas, err := ExecuteQueryByGorm(tableNames)
         if err !=nil {
            fmt.Println("queryByGorm err")
            return
         }
         bytesReturn, err := json.Marshal(datas)
         fmt.Println("results: ", bytesReturn)
         fmt.Println("results.len: ", len(bytesReturn))
         if query, ok := event.(*serf.Query); ok {
            if err := query.Respond(bytesReturn); err != nil {
               fmt.Printf("err: %s\n", err)
@@ -195,6 +209,18 @@
         //var res []*Rows
         //json.Unmarshal(bytesReturn, &res)
      }
   case serf.MemberEvent:
      if event.EventType() == serf.EventMemberLeave {
         if ev.Members !=nil && len(ev.Members) ==1 {
            leaveMember := ev.Members[0]
            leaveSql := "delete from cluster_node where node_id='"+leaveMember.Name+"'"
            ExecuteSqlByGorm([]string{ leaveSql })

            fmt.Println("EventMemberLeave,current Members:",ev.Members)
         }
         return
      }


   default:
      fmt.Printf("Unknown event type: %s\n", ev.EventType().String())
@@ -417,17 +443,18 @@

//GetDbFromCluster get the newest database after join cluster
//dbPathWrite the path where to write after got a database,
func (a *Agent) GetTableDataFromCluster(tableNames []string) error {
func (a *Agent) GetTableDataFromCluster(tableNames []string) (*[]string,error) {
   //members: get name of first member
   mbs := a.GroupMembers(a.conf.ClusterID)
   var specmembername string
   for _, m := range mbs {
      if m.Addr.String() != a.conf.BindAddr {
      fmt.Println("m",m)
      if m.Name != a.conf.NodeName {
         specmembername = m.Name
         break
      }
   }
   fmt.Println(specmembername)
   fmt.Println("mbs:",mbs,"a.conf.BindAddr:",a.conf.BindAddr,"specmembername:",specmembername)

   //query: get db file.
   params := serf.QueryParam{
@@ -441,27 +468,29 @@
   if err == nil || !strings.Contains(err.Error(), "cannot contain") {
      fmt.Println("err: ", err)
   }
   fmt.Println("Query.resp.err:",err,"resp:",resp)


   var wg sync.WaitGroup
   wg.Add(1)
   var dumpSqls []string
   go func() {
      defer wg.Done()
      respCh := resp.ResponseCh()
      for {
         select {
         case r := <-respCh:
            fmt.Println("x length is: ", len(r.Payload))
            // // byte to file.
            var dumpSqls []string
            fmt.Println("Query response's len:", len(r.Payload))
            err := json.Unmarshal(r.Payload, &dumpSqls)
            if err ==nil {
               if flag,_ := ExecuteSqlByGorm(dumpSqls);flag{
                  fmt.Println("data dump success")
               }
               fmt.Println("dumpSql:",dumpSqls)
               fmt.Println("data dump success")
            }
            return
         }
      }
   }()
   return nil
   wg.Wait()
   return &dumpSqls,nil

   //r, err = c.Query([]string{query}, false, false)
   //if err != nil {
@@ -476,24 +505,46 @@

}

type SqlUserEvent struct {
   Owner string `json:"owner"`
   Sql []string `json:"sql"`
}

//SyncSql boardcast sql to cluster
func (a *Agent) SyncSql(sqlOp string) {
func (a *Agent) SyncSql(sqlOp []string) {
   // event : use to send command to operate db.
   err := a.UserEvent("SyncSql", []byte(sqlOp), false)
   var sqlUe = SqlUserEvent{
      Owner: a.conf.NodeName,
      Sql: sqlOp,
   }
   ueB, err := json.Marshal(sqlUe)
   if err !=nil {
      fmt.Println("sqlUE marshal err:",err)
      return
   }
   err = a.UserEvent(UserEventSyncSql, ueB, false)
   if err == nil || !strings.Contains(err.Error(), "cannot contain") {
      fmt.Println("err: ", err)
   }
}

//更新同步库的比对缓存
func (a *Agent) SyncDbTablePersonCache(b []byte) {
   err := a.UserEvent(UserEventSyncDbTablePersonCache, b, false)
   if err !=nil{
      fmt.Println("UserEventSyncDbTablePersonCache err:",err)
   }
}

//Init serf Init
func Init(clusterID string, password string, nodeID string, ips []string) (*Agent, error) {
func Init(clusterID string, password string, nodeID string, addrs []string) (*Agent, error) {
   agent, err := InitNode(clusterID, password, nodeID)
   if err != nil {
      fmt.Printf("InitNode failed, error: %s", err)
      return agent, err
   }

   err = agent.JoinByNodeIP(ips)
   err = agent.JoinByNodeAddrs(addrs)
   if err != nil {
      fmt.Printf("JoinByNodeIP failed, error: %s", err)
      return agent, err
@@ -538,43 +589,49 @@
   return agent, nil
}

func (a *Agent) JoinByNodeIP(ips []string) error {
func (a *Agent) JoinByNodeAddrs(addrs []string) error {
   var nodes []string

   if len(ips) == 0 {
   if len(addrs) == 0 {
      return fmt.Errorf("No Nodes To Join!")
   }
   for _, ip := range ips {
      node := fmt.Sprintf("%s:%d", ip, DefaultBindPort)
      nodes = append(nodes, node)
   for _, addr := range addrs {
      nodes = append(nodes, addr)
   }

   n, err := a.Agent.Join(nodes, true)
   if err != nil || n == 0 {
      a.Stop()
      fmt.Println("Stop node")
      return fmt.Errorf("Error Encrypt Key!")
   }
   a.Agent.Join(nodes, true)

   return err
   return nil
}

type Node struct {
   clusterID string
   NodeID    string
   IP        string
   isAlive   int //StatusNone:0, StatusAlive:1, StatusLeaving:2, StatusLeft:3, StatusFailed:4
}
//func (a *Agent) JoinByNodeIP(ips []string) error {
//   var nodes []string
//
//   if len(ips) == 0 {
//      return fmt.Errorf("No Nodes To Join!")
//   }
//   for _, ip := range ips {
//      node := fmt.Sprintf("%s:%d", ip, DefaultBindPort)
//      nodes = append(nodes, node)
//   }
//
//   n, err := a.Agent.Join(nodes, true)
//   if err != nil || n == 0 {
//      return fmt.Errorf("Error Encrypt Key!")
//   }
//
//   return err
//}

func (a *Agent) GetNodes() (nodes []Node) {
   var node Node
func (a *Agent) GetNodes() (nodes []NodeInfo) {
   var node NodeInfo
   fmt.Println("a.conf.ClusterID:", a.conf.ClusterID)
   mbs := a.GroupMembers(a.conf.ClusterID)
   for _, mb := range mbs {
      node.NodeID = mb.Name
      node.IP = mb.Addr.String()
      node.isAlive = int(mb.Status)
      node.clusterID = mb.Tags[tagKeyClusterID]
      node.NodeAddress = mb.Addr.String() + ":" + strconv.Itoa(int(mb.Port))
      node.IsAlive = int(mb.Status)
      node.ClusterID = mb.Tags[tagKeyClusterID]

      nodes = append(nodes, node)
   }