ChenKaiLiuG/GoEdgeAPI
Archived
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

实现基础的智能调度

Browse Source
This commit is contained in:
刘祥超
2023-05-17 18:42:21 +08:00
parent 813ce44ceb
commit 6976454bde
33 changed files with 507 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/tasks/dns_task_executor.go
+109 -89
View File
@@ -87,8 +87,8 @@ func (this *DNSTaskExecutor) loop() error {
return err
}
}
case dnsmodels.DNSTaskTypeClusterChange:
err = this.doCluster(taskId, taskVersion, int64(task.ClusterId))
case dnsmodels.DNSTaskTypeClusterChange, dnsmodels.DNSTaskTypeClusterNodesChange:
err = this.doCluster(taskId, taskVersion, int64(task.ClusterId), task.Type == dnsmodels.DNSTaskTypeClusterNodesChange)
if err != nil {
err = dnsmodels.SharedDNSTaskDAO.UpdateDNSTaskError(nil, taskId, err.Error())
if err != nil {
@@ -300,7 +300,7 @@ func (this *DNSTaskExecutor) doNode(taskId int64, taskVersion int64, nodeCluster
// 转交给cluster统一处理
if nodeClusterId > 0 {
err = dnsmodels.SharedDNSTaskDAO.CreateClusterTask(tx, nodeClusterId, dnsmodels.DNSTaskTypeClusterChange)
err = dnsmodels.SharedDNSTaskDAO.CreateClusterTask(tx, nodeClusterId, dnsmodels.DNSTaskTypeClusterNodesChange)
if err != nil {
return err
}
@@ -310,7 +310,7 @@ func (this *DNSTaskExecutor) doNode(taskId int64, taskVersion int64, nodeCluster
return err
}
for _, clusterId := range clusterIds {
err = dnsmodels.SharedDNSTaskDAO.CreateClusterTask(tx, clusterId, dnsmodels.DNSTaskTypeClusterChange)
err = dnsmodels.SharedDNSTaskDAO.CreateClusterTask(tx, clusterId, dnsmodels.DNSTaskTypeClusterNodesChange)
if err != nil {
return err
}
@@ -323,8 +323,8 @@ func (this *DNSTaskExecutor) doNode(taskId int64, taskVersion int64, nodeCluster
}
// 修改集群相关记录
func (this *DNSTaskExecutor) doCluster(taskId int64, taskVersion int64, clusterId int64) error {
isOk := false
func (this *DNSTaskExecutor) doCluster(taskId int64, taskVersion int64, clusterId int64, nodesOnly bool) error {
var isOk = false
defer func() {
if isOk {
err := dnsmodels.SharedDNSTaskDAO.UpdateDNSTaskDone(nil, taskId, taskVersion)
@@ -378,6 +378,14 @@ func (this *DNSTaskExecutor) doCluster(taskId int64, taskVersion int64, clusterI
var isChanged = false
var addingNodeRecordKeysMap = map[string]bool{} // clusterDnsName_type_ip_route
for _, node := range nodes {
shouldSkip, shouldOverwrite, ipAddressesStrings, err := models.SharedNodeDAO.CheckNodeIPAddresses(tx, node)
if err != nil {
return err
}
if shouldSkip {
continue
}
routes, err := node.DNSRouteCodesForDomainId(domainId)
if err != nil {
return err
@@ -387,26 +395,36 @@ func (this *DNSTaskExecutor) doCluster(taskId int64, taskVersion int64, clusterI
}
// 所有的IP记录
ipAddresses, err := models.SharedNodeIPAddressDAO.FindAllEnabledAddressesWithNode(tx, int64(node.Id), nodeconfigs.NodeRoleNode)
if err != nil {
return err
if !shouldOverwrite {
ipAddresses, err := models.SharedNodeIPAddressDAO.FindAllEnabledAddressesWithNode(tx, int64(node.Id), nodeconfigs.NodeRoleNode)
if err != nil {
return err
}
if len(ipAddresses) == 0 {
continue
}
for _, ipAddress := range ipAddresses {
// 检查专属节点
if !ipAddress.IsValidInCluster(clusterId) {
continue
}
var ip = ipAddress.DNSIP()
if len(ip) == 0 || !ipAddress.CanAccess || !ipAddress.IsUp || !ipAddress.IsOn {
continue
}
if net.ParseIP(ip) == nil {
continue
}
ipAddressesStrings = append(ipAddressesStrings, ip)
}
}
if len(ipAddresses) == 0 {
if len(ipAddressesStrings) == 0 {
continue
}
for _, ipAddress := range ipAddresses {
// 检查专属节点
if !ipAddress.IsValidInCluster(clusterId) {
continue
}
var ip = ipAddress.DNSIP()
if len(ip) == 0 || !ipAddress.CanAccess || !ipAddress.IsUp || !ipAddress.IsOn {
continue
}
if net.ParseIP(ip) == nil {
continue
}
for _, ip := range ipAddressesStrings {
for _, route := range routes {
var key = route + "@" + ip
_, ok := oldRecordsMap[key]
@@ -456,80 +474,82 @@ func (this *DNSTaskExecutor) doCluster(taskId int64, taskVersion int64, clusterI
}
// 服务域名
servers, err := models.SharedServerDAO.FindAllServersDNSWithClusterId(tx, clusterId)
if err != nil {
return err
}
serverRecords := []*dnstypes.Record{} // 之所以用数组再存一遍，是因为dnsName可能会重复
serverRecordsMap := map[string]*dnstypes.Record{} // dnsName => *Record
for _, record := range records {
if record.Type == dnstypes.RecordTypeCNAME && record.Value == clusterDomain+"." {
serverRecords = append(serverRecords, record)
serverRecordsMap[record.Name] = record
if !nodesOnly {
servers, err := models.SharedServerDAO.FindAllServersDNSWithClusterId(tx, clusterId)
if err != nil {
return err
}
}
// 新增的域名
var serverDNSNames = []string{}
for _, server := range servers {
var dnsName = server.DnsName
if len(dnsName) == 0 {
continue
}
serverDNSNames = append(serverDNSNames, dnsName)
_, ok := serverRecordsMap[dnsName]
if !ok {
isChanged = true
err = manager.AddRecord(domain, &dnstypes.Record{
Id: "",
Name: dnsName,
Type: dnstypes.RecordTypeCNAME,
Value: clusterDomain + ".",
Route: "", // 注意这里为空，需要在执行过程中获取默认值
TTL: ttl,
})
if err != nil {
return err
serverRecords := []*dnstypes.Record{} // 之所以用数组再存一遍，是因为dnsName可能会重复
serverRecordsMap := map[string]*dnstypes.Record{} // dnsName => *Record
for _, record := range records {
if record.Type == dnstypes.RecordTypeCNAME && record.Value == clusterDomain+"." {
serverRecords = append(serverRecords, record)
serverRecordsMap[record.Name] = record
}
}
}
// 自动设置的CNAME
var cnameRecords = []string{}
if dnsConfig != nil {
cnameRecords = dnsConfig.CNAMERecords
}
for _, cnameRecord := range cnameRecords {
// 如果记录已存在，则跳过
if lists.ContainsString(serverDNSNames, cnameRecord) {
continue
}
serverDNSNames = append(serverDNSNames, cnameRecord)
_, ok := serverRecordsMap[cnameRecord]
if !ok {
isChanged = true
err = manager.AddRecord(domain, &dnstypes.Record{
Id: "",
Name: cnameRecord,
Type: dnstypes.RecordTypeCNAME,
Value: clusterDomain + ".",
Route: "", // 注意这里为空，需要在执行过程中获取默认值
TTL: ttl,
})
if err != nil {
return err
// 新增的域名
var serverDNSNames = []string{}
for _, server := range servers {
var dnsName = server.DnsName
if len(dnsName) == 0 {
continue
}
serverDNSNames = append(serverDNSNames, dnsName)
_, ok := serverRecordsMap[dnsName]
if !ok {
isChanged = true
err = manager.AddRecord(domain, &dnstypes.Record{
Id: "",
Name: dnsName,
Type: dnstypes.RecordTypeCNAME,
Value: clusterDomain + ".",
Route: "", // 注意这里为空，需要在执行过程中获取默认值
TTL: ttl,
})
if err != nil {
return err
}
}
}
}
// 多余的域名
for _, record := range serverRecords {
if !lists.ContainsString(serverDNSNames, record.Name) {
isChanged = true
err = manager.DeleteRecord(domain, record)
if err != nil {
return err
// 自动设置的CNAME
var cnameRecords = []string{}
if dnsConfig != nil {
cnameRecords = dnsConfig.CNAMERecords
}
for _, cnameRecord := range cnameRecords {
// 如果记录已存在，则跳过
if lists.ContainsString(serverDNSNames, cnameRecord) {
continue
}
serverDNSNames = append(serverDNSNames, cnameRecord)
_, ok := serverRecordsMap[cnameRecord]
if !ok {
isChanged = true
err = manager.AddRecord(domain, &dnstypes.Record{
Id: "",
Name: cnameRecord,
Type: dnstypes.RecordTypeCNAME,
Value: clusterDomain + ".",
Route: "", // 注意这里为空，需要在执行过程中获取默认值
TTL: ttl,
})
if err != nil {
return err
}
}
}
// 多余的域名
for _, record := range serverRecords {
if !lists.ContainsString(serverDNSNames, record.Name) {
isChanged = true
err = manager.DeleteRecord(domain, record)
if err != nil {
return err
}
}
}
}
internal/tasks/dns_task_executor_test.go
+2
View File
@@ -1,3 +1,5 @@
//go:build !plus
package tasks_test
import (
internal/tasks/health_check_cluster_task.go
+1 -1
View File
@@ -127,7 +127,7 @@ func (this *HealthCheckClusterTask) Loop() error {
if err != nil {
return err
}
var message = "有" + numberutils.FormatInt(len(failedResults)) + "个节点在健康检查中出现问题"
var message = "有" + numberutils.FormatInt(len(failedResults)) + "个节点IP在健康检查中出现问题"
err = models.NewMessageDAO().CreateClusterMessage(nil, nodeconfigs.NodeRoleNode, this.clusterId, models.MessageTypeHealthCheckFailed, models.MessageLevelError, message, message, failedResultsJSON)
if err != nil {
return err
internal/tasks/health_check_executor.go
+11 -1
View File
@@ -16,6 +16,7 @@ import (
"github.com/iwind/TeaGo/lists"
"github.com/iwind/TeaGo/maps"
"github.com/iwind/TeaGo/types"
timeutil "github.com/iwind/TeaGo/utils/time"
"net"
"net/http"
"strconv"
@@ -66,7 +67,7 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
var tx *dbs.Tx
for _, node := range nodes {
if !node.IsOn {
if !node.IsOn || node.IsBackupForCluster || node.IsBackupForGroup || (len(node.OfflineDay) > 0 && node.OfflineDay < timeutil.Format("Ymd")) {
continue
}
@@ -194,6 +195,15 @@ func (this *HealthCheckExecutor) runNode(healthCheckConfig *serverconfigs.Health
this.logErr("HealthCheckExecutor", err.Error())
return
}
// 触发节点动作
if !result.IsOk {
err := this.fireNodeActions(int64(result.Node.Id))
if err != nil {
this.logErr("HealthCheckExecutor", err.Error())
}
return
}
}
// 触发阈值
internal/tasks/health_check_executor_ext.go
+9
View File
@@ -0,0 +1,9 @@
// Copyright 2023 Liuxiangchao iwind.liu@gmail.com. All rights reserved. Official site: https://goedge.cn .
//go:build !plus
package tasks
// 触发节点动作
func (this *HealthCheckExecutor) fireNodeActions(nodeId int64) error {
return nil
}