实现节点自动切换到备用IP

internal/db/models/node_cluster_dao.go

@@ -792,6 +792,17 @@ func (this *NodeClusterDAO) FindEnabledNodeClustersWithIds(tx *dbs.Tx, clusterId
 	return
 }
 
+// ExistsEnabledCluster 检查集群是否存在
+func (this *NodeClusterDAO) ExistsEnabledCluster(tx *dbs.Tx, clusterId int64) (bool, error) {
+	if clusterId <= 0 {
+		return false, nil
+	}
+	return this.Query(tx).
+		Pk(clusterId).
+		State(NodeClusterStateEnabled).
+		Exist()
+}
+
 // NotifyUpdate 通知更新
 func (this *NodeClusterDAO) NotifyUpdate(tx *dbs.Tx, clusterId int64) error {
 	return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)

internal/db/models/node_ip_address_dao.go

@@ -401,6 +401,22 @@ func (this *NodeIPAddressDAO) UpdateAddressIsUp(tx *dbs.Tx, addressId int64, isU
 	return this.NotifyUpdate(tx, addressId)
 }
 
+// UpdateAddressBackupIP 设置备用IP
+func (this *NodeIPAddressDAO) UpdateAddressBackupIP(tx *dbs.Tx, addressId int64, thresholdId int64, ip string) error {
+	if addressId <= 0 {
+		return errors.New("invalid addressId")
+	}
+	var op = NewNodeIPAddressOperator()
+	op.Id = addressId
+	op.BackupThresholdId = thresholdId
+	op.BackupIP = ip
+	err := this.Save(tx, op)
+	if err != nil {
+		return err
+	}
+	return this.NotifyUpdate(tx, addressId)
+}
+
 // NotifyUpdate 通知更新
 func (this *NodeIPAddressDAO) NotifyUpdate(tx *dbs.Tx, addressId int64) error {
 	address, err := this.Query(tx).

internal/db/models/node_ip_address_dao_test.go

@@ -29,3 +29,14 @@ func TestNodeIPAddressDAO_FireThresholds(t *testing.T) {
 	}
 	t.Log("ok")
 }
+
+func TestNodeIPAddressDAO_LoopTasks(t *testing.T) {
+	dbs.NotifyReady()
+
+	var tx *dbs.Tx
+	err := SharedNodeIPAddressDAO.loopTask(tx, nodeconfigs.NodeRoleNode)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok")
+}

internal/db/models/node_ip_address_log_dao.go

@@ -45,6 +45,7 @@ func (this *NodeIPAddressLogDAO) CreateLog(tx *dbs.Tx, adminId int64, addrId int
 	op.CanAccess = addr.CanAccess
 	op.IsOn = addr.IsOn
 	op.IsUp = addr.IsUp
+	op.BackupIP = addr.BackupIP
 	op.Day = timeutil.Format("Ymd")
 	return this.Save(tx, op)
 }

internal/db/models/node_ip_address_log_model.go

@@ -11,6 +11,7 @@ type NodeIPAddressLog struct {
 	IsOn        uint8  `field:"isOn"`        // 是否启用
 	CanAccess   uint8  `field:"canAccess"`   // 是否可访问
 	Day         string `field:"day"`         // YYYYMMDD，用来清理
+	BackupIP    string `field:"backupIP"`    // 备用IP
 }
 
 type NodeIPAddressLogOperator struct {
@@ -23,6 +24,7 @@ type NodeIPAddressLogOperator struct {
 	IsOn        interface{} // 是否启用
 	CanAccess   interface{} // 是否可访问
 	Day         interface{} // YYYYMMDD，用来清理
+	BackupIP    interface{} // 备用IP
 }
 
 func NewNodeIPAddressLogOperator() *NodeIPAddressLogOperator {

internal/db/models/node_ip_address_model.go

@@ -2,35 +2,39 @@ package models
 
 // NodeIPAddress 节点IP地址
 type NodeIPAddress struct {
-	Id          uint32 `field:"id"`          // ID
+	Id                uint32 `field:"id"`                // ID
-	NodeId      uint32 `field:"nodeId"`      // 节点ID
+	NodeId            uint32 `field:"nodeId"`            // 节点ID
-	Role        string `field:"role"`        // 节点角色
+	Role              string `field:"role"`              // 节点角色
-	Name        string `field:"name"`        // 名称
+	Name              string `field:"name"`              // 名称
-	Ip          string `field:"ip"`          // IP地址
+	Ip                string `field:"ip"`                // IP地址
-	Description string `field:"description"` // 描述
+	Description       string `field:"description"`       // 描述
-	State       uint8  `field:"state"`       // 状态
+	State             uint8  `field:"state"`             // 状态
-	Order       uint32 `field:"order"`       // 排序
+	Order             uint32 `field:"order"`             // 排序
-	CanAccess   uint8  `field:"canAccess"`   // 是否可以访问
+	CanAccess         uint8  `field:"canAccess"`         // 是否可以访问
-	IsOn        uint8  `field:"isOn"`        // 是否启用
+	IsOn              uint8  `field:"isOn"`              // 是否启用
-	IsUp        uint8  `field:"isUp"`        // 是否上线
+	IsUp              uint8  `field:"isUp"`              // 是否上线
-	//Thresholds   string `field:"thresholds"`   // 上线阈值
+	Thresholds        string `field:"thresholds"`        // 上线阈值
-	Connectivity string `field:"connectivity"` // 连通性状态
+	Connectivity      string `field:"connectivity"`      // 连通性状态
+	BackupIP          string `field:"backupIP"`          // 备用IP
+	BackupThresholdId uint32 `field:"backupThresholdId"` // 触发备用IP的阈值
 }
 
 type NodeIPAddressOperator struct {
-	Id          interface{} // ID
+	Id                interface{} // ID
-	NodeId      interface{} // 节点ID
+	NodeId            interface{} // 节点ID
-	Role        interface{} // 节点角色
+	Role              interface{} // 节点角色
-	Name        interface{} // 名称
+	Name              interface{} // 名称
-	Ip          interface{} // IP地址
+	Ip                interface{} // IP地址
-	Description interface{} // 描述
+	Description       interface{} // 描述
-	State       interface{} // 状态
+	State             interface{} // 状态
-	Order       interface{} // 排序
+	Order             interface{} // 排序
-	CanAccess   interface{} // 是否可以访问
+	CanAccess         interface{} // 是否可以访问
-	IsOn        interface{} // 是否启用
+	IsOn              interface{} // 是否启用
-	IsUp        interface{} // 是否上线
+	IsUp              interface{} // 是否上线
-	//Thresholds   interface{} // 上线阈值
+	Thresholds        interface{} // 上线阈值
-	Connectivity interface{} // 连通性状态
+	Connectivity      interface{} // 连通性状态
+	BackupIP          interface{} // 备用IP
+	BackupThresholdId interface{} // 触发备用IP的阈值
 }
 
 func NewNodeIPAddressOperator() *NodeIPAddressOperator {

internal/db/models/node_ip_address_model_ext.go

@@ -6,6 +6,7 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 )
 
+// DecodeConnectivity 解析联通数值
 func (this *NodeIPAddress) DecodeConnectivity() *nodeconfigs.Connectivity {
 	var connectivity = &nodeconfigs.Connectivity{}
 	if len(this.Connectivity) > 0 {
@@ -16,3 +17,28 @@ func (this *NodeIPAddress) DecodeConnectivity() *nodeconfigs.Connectivity {
 	}
 	return connectivity
 }
+
+// DNSIP 获取当前DNS可以使用的IP
+func (this *NodeIPAddress) DNSIP() string {
+	var backupIP = this.DecodeBackupIP()
+	if len(backupIP) > 0 {
+		return backupIP
+	}
+	return this.Ip
+}
+
+// DecodeBackupIP 获取备用IP
+func (this *NodeIPAddress) DecodeBackupIP() string {
+	if this.BackupThresholdId > 0 && len(this.BackupIP) > 0 {
+		// 阈值是否存在
+		b, err := SharedNodeIPAddressThresholdDAO.ExistsEnabledThreshold(nil, int64(this.BackupThresholdId))
+		if err != nil {
+			remotelogs.Error("NodeIPAddress.DNSIP", "check enabled threshold failed: "+err.Error())
+		} else {
+			if b {
+				return this.BackupIP
+			}
+		}
+	}
+	return ""
+}

internal/db/models/node_ip_address_threshold_dao.go

@@ -234,3 +234,11 @@ func (this *NodeIPAddressThresholdDAO) formatThreshold(tx *dbs.Tx, threshold *No
 
 	return nil
 }
+
+// ExistsEnabledThreshold 检查阈值是否可以使用
+func (this *NodeIPAddressThresholdDAO) ExistsEnabledThreshold(tx *dbs.Tx, thresholdId int64) (bool, error) {
+	return this.Query(tx).
+		Pk(thresholdId).
+		State(NodeIPAddressThresholdStateEnabled).
+		Exist()
+}

internal/rpc/services/service_dns_domain.go

@@ -441,7 +441,7 @@ func (this *DNSDomainService) findClusterDNSChanges(cluster *models.NodeCluster,
 		}
 		for _, route := range routeCodes {
 			for _, ipAddress := range ipAddresses {
-				ip := ipAddress.Ip
+				ip := ipAddress.DNSIP()
 				if len(ip) == 0 {
 					continue
 				}

internal/rpc/services/service_node.go

@@ -1262,7 +1262,7 @@ func (this *NodeService) FindAllEnabledNodesDNSWithNodeClusterId(ctx context.Con
 		}
 
 		for _, ipAddress := range ipAddresses {
-			ip := ipAddress.Ip
+			ip := ipAddress.DNSIP()
 			if len(ip) == 0 {
 				continue
 			}

internal/rpc/services/service_node_ip_address.go

@@ -130,6 +130,7 @@ func (this *NodeIPAddressService) FindEnabledNodeIPAddress(ctx context.Context,
 			CanAccess:   address.CanAccess == 1,
 			IsOn:        address.IsOn == 1,
 			IsUp:        address.IsUp == 1,
+			BackupIP:    address.DecodeBackupIP(),
 		}
 	}
 
@@ -165,6 +166,7 @@ func (this *NodeIPAddressService) FindAllEnabledIPAddressesWithNodeId(ctx contex
 			CanAccess:   address.CanAccess == 1,
 			IsOn:        address.IsOn == 1,
 			IsUp:        address.IsUp == 1,
+			BackupIP:    address.DecodeBackupIP(),
 		})
 	}
 
@@ -214,6 +216,7 @@ func (this *NodeIPAddressService) ListEnabledIPAddresses(ctx context.Context, re
 			CanAccess:   addr.CanAccess == 1,
 			IsOn:        addr.IsOn == 1,
 			IsUp:        addr.IsUp == 1,
+			BackupIP:    addr.DecodeBackupIP(),
 		})
 	}
 	return &pb.ListEnabledIPAddressesResponse{NodeIPAddresses: pbAddrs}, nil

internal/rpc/services/service_node_ip_address_log.go

@@ -81,6 +81,7 @@ func (this *NodeIPAddressLogService) ListNodeIPAddressLogs(ctx context.Context,
 			IsOn:          log.IsOn == 1,
 			IsUp:          log.IsUp == 1,
 			CanAccess:     log.CanAccess == 1,
+			BackupIP:      log.BackupIP,
 			NodeIPAddress: pbAddr,
 			Admin:         pbAdmin,
 		})

internal/rpc/services/service_node_value.go

@@ -4,7 +4,6 @@ package services
 
 import (
 	"context"
-	teaconst "github.com/TeaOSLab/EdgeAPI/internal/const"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
@@ -46,15 +45,6 @@ func (this *NodeValueService) CreateNodeValue(ctx context.Context, req *pb.Creat
 		return nil, err
 	}
 
-	// 触发IP阈值
-	// 企业版专有
-	if teaconst.IsPlus {
-		err = models.SharedNodeIPAddressDAO.FireThresholds(tx, role, nodeId)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	return this.Success()
 }
 

internal/tasks/dns_task_executor.go

@@ -311,7 +311,7 @@ func (this *DNSTaskExecutor) doCluster(taskId int64, clusterId int64) error {
 			continue
 		}
 		for _, ipAddress := range ipAddresses {
-			ip := ipAddress.Ip
+			ip := ipAddress.DNSIP()
 			if len(ip) == 0 || ipAddress.CanAccess == 0 || ipAddress.IsUp == 0 || ipAddress.IsOn == 0 {
 				continue
 			}