添加主节点变更和主动切换功能

zhangzengfei

2023-10-20 71b8885babe6dfd25c91b007018347c0c1bfac74

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
package models
 
import (
    "basic.com/valib/logger.git"
)
 
const (
    IsAlive_StatusNone    = 0
    IsAlive_StatusAlive   = 1
    IsAlive_StatusLeaving = 2
    IsAlive_StatusLeft    = 3
    IsAlive_StatusFailed  = 4
)
 
type Node struct {
    Id         string `gorm:"primary_key;column:id" json:"id"`
    ClusterId  string `gorm:"column:cluster_id" json:"cluster_id"`
    NodeName   string `gorm:"column:node_name" json:"node_name"`
    NodeId     string `gorm:"column:node_id" json:"node_id"`
    NodeIp     string `gorm:"column:node_ip" json:"node_ip"`
    CreateTime string `gorm:"column:create_time" json:"create_time"`
    IsDelete   bool   `gorm:"column:isDelete;default:0" json:"isDelete"`
    DeviceType string `gorm:"column:device_type" json:"device_type"` //设备型号`
    DriftState string `gorm:"column:drift_state" json:"drift_state"` //漂移状态, master,backup
    Online     bool   `gorm:"column:online;default:1" json:"online"` //在线状态
}
 
func (Node) TableName() string {
    return "cluster_node"
}
 
func (n *Node) FindIpByNode(nodeId string) (string, error) {
    node := Node{}
    if err := db.Raw("select id, node_ip from cluster_node where isDelete=0 and id = '" + nodeId + "'").First(&node).Error; err != nil {
        return "", err
    }
    return node.NodeIp, nil
}
 
func (n *Node) FindNodesByClusterId(clusterId string) (nodes []Node, err error) {
    if err = db.Raw("select * from cluster_node where isDelete=0 and cluster_id=?", clusterId).Scan(&nodes).Error; err != nil {
        return nil, err
    }
    return nodes, nil
}
 
func (n *Node) FindNodes() (nodes []Node, err error) {
    if err = db.Raw("select * from cluster_node").Scan(&nodes).Error; err != nil {
        return nil, err
    }
    return nodes, nil
}
 
func (n *Node) FindFirstNode() (node Node, err error) {
    if err = db.Raw("select * from cluster_node where isDelete=0 order by create_time limit 1").Scan(&node).Error; err != nil {
        return node, err
    }
    return node, nil
}
 
func (n *Node) FindNodeById(id string) (node Node, err error) {
    if err = db.Raw("select * from cluster_node where id=?", id).Scan(&node).Error; err != nil {
        return node, err
    }
    return node, nil
}
 
func (n *Node) UpdateDriftStateByNodeId(driftState string, nodeId string, sync bool) bool {
    var err error
    tx := GetDB().Begin()
    if !sync {
        GetDB().LogMode(false)
        defer db.LogMode(true)
    }
 
    defer func() {
        if err != nil && tx != nil {
            logger.Error("updateDriftState err:", err)
            tx.Rollback()
        }
    }()
    if driftState == "master" {
        if err = tx.Exec("update "+n.TableName()+" set drift_state=? where id=?", driftState, nodeId).Error; err != nil {
            return false
        }
        if err = tx.Exec("update "+n.TableName()+" set drift_state='slave' where id !=?", nodeId).Error; err != nil {
            return false
        }
    } else {
        if err = tx.Exec("update "+n.TableName()+" set drift_state=? where id=?", driftState, nodeId).Error; err != nil {
            return false
        }
    }
 
    tx.Commit()
    return true
}