增加DNS同步任务状态显示

internal/rpc/services/service_dns_domain.go

@@ -12,6 +12,7 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/iwind/TeaGo/lists"
 	"github.com/iwind/TeaGo/maps"
+	"net"
 )
 
 // DNS域名相关服务
@@ -410,11 +411,11 @@ func (this *DNSDomainService) findClusterDNSChanges(cluster *models.NodeCluster,
 	// 新增的节点域名
 	nodeKeys := []string{}
 	for _, node := range nodes {
-		ipAddr, err := models.SharedNodeIPAddressDAO.FindFirstNodeIPAddress(tx, int64(node.Id))
+		ipAddresses, err := models.SharedNodeIPAddressDAO.FindNodeAccessIPAddresses(tx, int64(node.Id))
 		if err != nil {
 			return nil, nil, nil, 0, 0, false, false, err
 		}
-		if len(ipAddr) == 0 {
+		if len(ipAddresses) == 0 {
 			continue
 		}
 		routeCodes, err := node.DNSRouteCodesForDomainId(int64(cluster.DnsDomainId))
@@ -425,23 +426,32 @@ func (this *DNSDomainService) findClusterDNSChanges(cluster *models.NodeCluster,
 			continue
 		}
 		for _, route := range routeCodes {
-			key := ipAddr + "_" + route
-			nodeKeys = append(nodeKeys, key)
-			record, ok := nodeRecordMapping[key]
-			if !ok {
-				result = append(result, maps.Map{
-					"action": "create",
-					"record": &dnsclients.Record{
-						Id:    "",
-						Name:  clusterDnsName,
-						Type:  dnsclients.RecordTypeA,
-						Value: ipAddr,
-						Route: route,
-					},
-				})
-				nodesChanged = true
-			} else {
-				doneNodeRecords = append(doneNodeRecords, record)
+			for _, ipAddress := range ipAddresses {
+				ip := ipAddress.Ip
+				if len(ip) == 0 {
+					continue
+				}
+				if net.ParseIP(ip) == nil {
+					continue
+				}
+				key := ip + "_" + route
+				nodeKeys = append(nodeKeys, key)
+				record, ok := nodeRecordMapping[key]
+				if !ok {
+					result = append(result, maps.Map{
+						"action": "create",
+						"record": &dnsclients.Record{
+							Id:    "",
+							Name:  clusterDnsName,
+							Type:  dnsclients.RecordTypeA,
+							Value: ip,
+							Route: route,
+						},
+					})
+					nodesChanged = true
+				} else {
+					doneNodeRecords = append(doneNodeRecords, record)
+				}
 			}
 		}
 	}

internal/rpc/services/service_dns_task.go

@@ -0,0 +1,110 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models/dns"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+)
+
+// DNS同步相关任务
+type DNSTaskService struct {
+	BaseService
+}
+
+// 检查是否有正在执行的任务
+func (this *DNSTaskService) ExistsDNSTasks(ctx context.Context, req *pb.ExistsDNSTasksRequest) (*pb.ExistsDNSTasksResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	existDoingTasks, err := dns.SharedDNSTaskDAO.ExistDoingTasks(tx)
+	if err != nil {
+		return nil, err
+	}
+
+	existErrorTasks, err := dns.SharedDNSTaskDAO.ExistErrorTasks(tx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.ExistsDNSTasksResponse{
+		ExistTasks: existDoingTasks,
+		ExistError: existErrorTasks,
+	}, nil
+}
+
+// 查找正在执行的所有任务
+func (this *DNSTaskService) FindAllDoingDNSTasks(ctx context.Context, req *pb.FindAllDoingDNSTasksRequest) (*pb.FindAllDoingDNSTasksResponse, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	var tx = this.NullTx()
+	tasks, err := dns.SharedDNSTaskDAO.FindAllDoingOrErrorTasks(tx)
+	if err != nil {
+		return nil, err
+	}
+
+	pbTasks := []*pb.DNSTask{}
+	for _, task := range tasks {
+		pbTask := &pb.DNSTask{
+			Id:        int64(task.Id),
+			Type:      task.Type,
+			IsDone:    task.IsDone == 1,
+			IsOk:      task.IsOk == 1,
+			Error:     task.Error,
+			UpdatedAt: int64(task.UpdatedAt),
+		}
+
+		switch task.Type {
+		case dns.DNSTaskTypeClusterChange:
+			clusterName, err := models.SharedNodeClusterDAO.FindNodeClusterName(tx, int64(task.ClusterId))
+			if err != nil {
+				return nil, err
+			}
+			if len(clusterName) == 0 {
+				clusterName = "集群[" + fmt.Sprintf("%d", task.ClusterId) + "]"
+			}
+			pbTask.NodeCluster = &pb.NodeCluster{Id: int64(task.ClusterId), Name: clusterName}
+		case dns.DNSTaskTypeNodeChange:
+			nodeName, err := models.SharedNodeDAO.FindNodeName(tx, int64(task.NodeId))
+			if err != nil {
+				return nil, err
+			}
+			if len(nodeName) == 0 {
+				nodeName = "节点[" + fmt.Sprintf("%d", task.NodeId) + "]"
+			}
+			pbTask.Node = &pb.Node{Id: int64(task.NodeId), Name: nodeName}
+		case dns.DNSTaskTypeServerChange:
+			serverName, err := models.SharedServerDAO.FindEnabledServerName(tx, int64(task.ServerId))
+			if err != nil {
+				return nil, err
+			}
+			if len(serverName) == 0 {
+				serverName = "服务[" + fmt.Sprintf("%d", task.ServerId) + "]"
+			}
+			pbTask.Server = &pb.Server{Id: int64(task.ServerId), Name: serverName}
+		}
+		pbTasks = append(pbTasks, pbTask)
+	}
+	return &pb.FindAllDoingDNSTasksResponse{DnsTasks: pbTasks}, nil
+}
+
+// 删除任务
+func (this *DNSTaskService) DeleteDNSTask(ctx context.Context, req *pb.DeleteDNSTaskRequest) (*pb.RPCSuccess, error) {
+	_, err := this.ValidateAdmin(ctx, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	err = dns.SharedDNSTaskDAO.DeleteDNSTask(this.NullTx(), req.DnsTaskId)
+	if err != nil {
+		return nil, err
+	}
+	return this.Success()
+}

internal/rpc/services/service_node.go

@@ -7,31 +7,16 @@ import (
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models/dns"
 	"github.com/TeaOSLab/EdgeAPI/internal/dnsclients"
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
-	"github.com/TeaOSLab/EdgeAPI/internal/events"
 	"github.com/TeaOSLab/EdgeAPI/internal/installers"
 	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
-	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/logs"
 	"github.com/iwind/TeaGo/types"
+	"net"
 )
 
-func init() {
-	dbs.OnReady(func() {
-		go func() {
-			service := &NodeService{}
-			for nodeId := range events.NodeDNSChanges {
-				err := service.notifyNodeDNSChanged(nodeId)
-				if err != nil {
-					logs.Println("[ERROR]change node dns: " + err.Error())
-				}
-			}
-		}()
-	})
-}
-
 // 边缘节点相关服务
 type NodeService struct {
 	BaseService
@@ -69,14 +54,6 @@ func (this *NodeService) CreateNode(ctx context.Context, req *pb.CreateNodeReque
 		}
 	}
 
-	// 同步DNS
-	go func() {
-		err := this.notifyNodeDNSChanged(nodeId)
-		if err != nil {
-			logs.Println("sync node DNS error: " + err.Error())
-		}
-	}()
-
 	return &pb.CreateNodeResponse{
 		NodeId: nodeId,
 	}, nil
@@ -344,14 +321,6 @@ func (this *NodeService) DeleteNode(ctx context.Context, req *pb.DeleteNodeReque
 		return nil, err
 	}
 
-	// 同步DNS
-	go func() {
-		err := this.notifyNodeDNSChanged(req.NodeId)
-		if err != nil {
-			logs.Println("sync node DNS error: " + err.Error())
-		}
-	}()
-
 	// 删除节点相关任务
 	err = models.SharedNodeTaskDAO.DeleteNodeTasks(tx, req.NodeId)
 	if err != nil {
@@ -404,15 +373,6 @@ func (this *NodeService) UpdateNode(ctx context.Context, req *pb.UpdateNodeReque
 		}
 	}
 
-	// 同步DNS
-	go func() {
-		// TODO 只有状态变化的时候才需要同步
-		err := this.notifyNodeDNSChanged(req.NodeId)
-		if err != nil {
-			logs.Println("sync node DNS error: " + err.Error())
-		}
-	}()
-
 	return this.Success()
 }
 
@@ -675,14 +635,6 @@ func (this *NodeService) StartNode(ctx context.Context, req *pb.StartNodeRequest
 		}, nil
 	}
 
-	// 同步DNS
-	go func() {
-		err := this.notifyNodeDNSChanged(req.NodeId)
-		if err != nil {
-			logs.Println("sync node DNS error: " + err.Error())
-		}
-	}()
-
 	return &pb.StartNodeResponse{IsOk: true}, nil
 }
 
@@ -702,14 +654,6 @@ func (this *NodeService) StopNode(ctx context.Context, req *pb.StopNodeRequest)
 		}, nil
 	}
 
-	// 同步DNS
-	go func() {
-		err := this.notifyNodeDNSChanged(req.NodeId)
-		if err != nil {
-			logs.Println("sync node DNS error: " + err.Error())
-		}
-	}()
-
 	return &pb.StopNodeResponse{IsOk: true}, nil
 }
 
@@ -1098,7 +1042,7 @@ func (this *NodeService) FindAllEnabledNodesDNSWithClusterId(ctx context.Context
 	}
 	result := []*pb.NodeDNSInfo{}
 	for _, node := range nodes {
-		ipAddr, err := models.SharedNodeIPAddressDAO.FindFirstNodeIPAddress(tx, int64(node.Id))
+		ipAddresses, err := models.SharedNodeIPAddressDAO.FindNodeAccessIPAddresses(tx, int64(node.Id))
 		if err != nil {
 			return nil, err
 		}
@@ -1121,12 +1065,21 @@ func (this *NodeService) FindAllEnabledNodesDNSWithClusterId(ctx context.Context
 			}
 		}
 
-		result = append(result, &pb.NodeDNSInfo{
-			Id:     int64(node.Id),
-			Name:   node.Name,
-			IpAddr: ipAddr,
-			Routes: pbRoutes,
-		})
+		for _, ipAddress := range ipAddresses {
+			ip := ipAddress.Ip
+			if len(ip) == 0 {
+				continue
+			}
+			if net.ParseIP(ip) == nil {
+				continue
+			}
+			result = append(result, &pb.NodeDNSInfo{
+				Id:     int64(node.Id),
+				Name:   node.Name,
+				IpAddr: ip,
+				Routes: pbRoutes,
+			})
+		}
 	}
 	return &pb.FindAllEnabledNodesDNSWithClusterIdResponse{Nodes: result}, nil
 }
@@ -1150,7 +1103,7 @@ func (this *NodeService) FindEnabledNodeDNS(ctx context.Context, req *pb.FindEna
 		return &pb.FindEnabledNodeDNSResponse{Node: nil}, nil
 	}
 
-	ipAddr, err := models.SharedNodeIPAddressDAO.FindFirstNodeIPAddress(tx, int64(node.Id))
+	ipAddr, err := models.SharedNodeIPAddressDAO.FindFirstNodeAccessIPAddress(tx, int64(node.Id))
 	if err != nil {
 		return nil, err
 	}
@@ -1237,7 +1190,7 @@ func (this *NodeService) UpdateNodeDNS(ctx context.Context, req *pb.UpdateNodeDN
 
 	// 修改IP
 	if len(req.IpAddr) > 0 {
-		ipAddrId, err := models.SharedNodeIPAddressDAO.FindFirstNodeIPAddressId(tx, req.NodeId)
+		ipAddrId, err := models.SharedNodeIPAddressDAO.FindFirstNodeAccessIPAddressId(tx, req.NodeId)
 		if err != nil {
 			return nil, err
 		}
@@ -1257,50 +1210,6 @@ func (this *NodeService) UpdateNodeDNS(ctx context.Context, req *pb.UpdateNodeDN
 	return this.Success()
 }
 
-// 自动同步DNS状态
-func (this *NodeService) notifyNodeDNSChanged(nodeId int64) error {
-	tx := this.NullTx()
-
-	clusterId, err := models.SharedNodeDAO.FindNodeClusterId(tx, nodeId)
-	if err != nil {
-		return err
-	}
-	dnsInfo, err := models.SharedNodeClusterDAO.FindClusterDNSInfo(tx, clusterId)
-	if err != nil {
-		return err
-	}
-	if dnsInfo == nil {
-		return nil
-	}
-	if len(dnsInfo.DnsName) == 0 || dnsInfo.DnsDomainId == 0 {
-		return nil
-	}
-	dnsConfig, err := dnsInfo.DecodeDNSConfig()
-	if err != nil {
-		return err
-	}
-	if !dnsConfig.NodesAutoSync {
-		return nil
-	}
-
-	// 执行同步
-	domainService := &DNSDomainService{}
-	resp, err := domainService.syncClusterDNS(&pb.SyncDNSDomainDataRequest{
-		DnsDomainId:   int64(dnsInfo.DnsDomainId),
-		NodeClusterId: clusterId,
-	})
-	if err != nil {
-		return err
-	}
-	if !resp.IsOk {
-		err = models.SharedMessageDAO.CreateClusterMessage(tx, clusterId, models.MessageTypeClusterDNSSyncFailed, models.LevelError, "集群DNS同步失败："+resp.Error, nil)
-		if err != nil {
-			logs.Println("[NODE_SERVICE]" + err.Error())
-		}
-	}
-	return nil
-}
-
 // 计算某个区域下的节点数量
 func (this *NodeService) CountAllEnabledNodesWithNodeRegionId(ctx context.Context, req *pb.CountAllEnabledNodesWithNodeRegionIdRequest) (*pb.RPCCountResponse, error) {
 	_, err := this.ValidateAdmin(ctx, 0)

internal/rpc/services/service_server.go

@@ -11,7 +11,6 @@ import (
 	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
-	"github.com/iwind/TeaGo/logs"
 	"github.com/iwind/TeaGo/maps"
 	"github.com/iwind/TeaGo/types"
 	timeutil "github.com/iwind/TeaGo/utils/time"
@@ -119,17 +118,6 @@ func (this *ServerService) UpdateServerBasic(ctx context.Context, req *pb.Update
 		return nil, err
 	}
 
-	// 检查服务变化
-	oldIsOn := server.IsOn == 1
-	if oldIsOn != req.IsOn {
-		go func() {
-			err := this.notifyServerDNSChanged(req.ServerId)
-			if err != nil {
-				logs.Println("[DNS]notify server changed: " + err.Error())
-			}
-		}()
-	}
-
 	return this.Success()
 }
 
@@ -373,7 +361,7 @@ func (this *ServerService) FindServerNames(ctx context.Context, req *pb.FindServ
 		}
 	}
 
-	serverNamesJSON, isAuditing, auditingServerNamesJSON, auditingResultJSON, err := models.SharedServerDAO.FindServerNames(tx, req.ServerId)
+	serverNamesJSON, isAuditing, auditingServerNamesJSON, auditingResultJSON, err := models.SharedServerDAO.FindServerServerNames(tx, req.ServerId)
 	if err != nil {
 		return nil, err
 	}
@@ -474,14 +462,6 @@ func (this *ServerService) UpdateServerNamesAuditing(ctx context.Context, req *p
 		}
 	}
 
-	// 通知服务更新
-	go func() {
-		err := this.notifyServerDNSChanged(req.ServerId)
-		if err != nil {
-			logs.Println("[DNS]notify server changed: " + err.Error())
-		}
-	}()
-
 	return this.Success()
 }
 
@@ -1095,50 +1075,6 @@ func (this *ServerService) FindEnabledServerDNS(ctx context.Context, req *pb.Fin
 	}, nil
 }
 
-// 自动同步DNS状态
-func (this *ServerService) notifyServerDNSChanged(serverId int64) error {
-	tx := this.NullTx()
-
-	clusterId, err := models.SharedServerDAO.FindServerClusterId(tx, serverId)
-	if err != nil {
-		return err
-	}
-	dnsInfo, err := models.SharedNodeClusterDAO.FindClusterDNSInfo(tx, clusterId)
-	if err != nil {
-		return err
-	}
-	if dnsInfo == nil {
-		return nil
-	}
-	if len(dnsInfo.DnsName) == 0 || dnsInfo.DnsDomainId == 0 {
-		return nil
-	}
-	dnsConfig, err := dnsInfo.DecodeDNSConfig()
-	if err != nil {
-		return err
-	}
-	if !dnsConfig.ServersAutoSync {
-		return nil
-	}
-
-	// 执行同步
-	domainService := &DNSDomainService{}
-	resp, err := domainService.syncClusterDNS(&pb.SyncDNSDomainDataRequest{
-		DnsDomainId:   int64(dnsInfo.DnsDomainId),
-		NodeClusterId: clusterId,
-	})
-	if err != nil {
-		return err
-	}
-	if !resp.IsOk {
-		err = models.SharedMessageDAO.CreateClusterMessage(tx, clusterId, models.MessageTypeClusterDNSSyncFailed, models.LevelError, "集群DNS同步失败："+resp.Error, nil)
-		if err != nil {
-			logs.Println("[NODE_SERVICE]" + err.Error())
-		}
-	}
-	return nil
-}
-
 // 检查服务是否属于某个用户
 func (this *ServerService) CheckUserServer(ctx context.Context, req *pb.CheckUserServerRequest) (*pb.RPCSuccess, error) {
 	userId, err := this.ValidateUser(ctx)