TRKJ/EdgeAPI
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeAPI.git synced 2025-11-07 02:20:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

当服务配置变化时创建单个服务通知任务

Browse Source
This commit is contained in:
刘祥超
2022-01-19 22:15:01 +08:00
parent 5205136809
commit ef045e90f2
16 changed files with 60 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
internal/db/models/ip_item_dao.go
View File

@@ -464,7 +464,7 @@ func (this *IPItemDAO) NotifyUpdate(tx *dbs.Tx, itemId int64) error {
return err
}
for _, clusterId := range clusterIds {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeIPItemChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeIPItemChanged)
if err != nil {
return err
}
@@ -472,7 +472,7 @@ func (this *IPItemDAO) NotifyUpdate(tx *dbs.Tx, itemId int64) error {
} else {
clusterIds, err := SharedNodeClusterDAO.FindAllEnabledNodeClusterIds(tx)
for _, clusterId := range clusterIds {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeIPItemChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeIPItemChanged)
if err != nil {
return err
}
@@ -524,7 +524,7 @@ func (this *IPItemDAO) NotifyUpdate(tx *dbs.Tx, itemId int64) error {
if len(resultClusterIds) > 0 {
for _, clusterId := range resultClusterIds {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeIPItemChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeIPItemChanged)
if err != nil {
return err
}

2
internal/db/models/ip_list_dao.go
View File

@@ -284,7 +284,7 @@ func (this *IPListDAO) NotifyUpdate(tx *dbs.Tx, listId int64, taskType NodeTaskT
if len(resultClusterIds) > 0 {
for _, clusterId := range resultClusterIds {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, taskType)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, taskType)
if err != nil {
return err
}

4
internal/db/models/metric_item_dao.go
View File

@@ -334,7 +334,7 @@ func (this *MetricItemDAO) NotifyUpdate(tx *dbs.Tx, itemId int64, isPublic bool)
return err
}
for _, clusterId := range clusterIds {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeConfigChanged)
if err != nil {
return err
}
@@ -346,7 +346,7 @@ func (this *MetricItemDAO) NotifyUpdate(tx *dbs.Tx, itemId int64, isPublic bool)
return err
}
for _, clusterId := range clusterIds {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeConfigChanged)
if err != nil {
return err
}

2
internal/db/models/nameservers/ns_domain_dao.go
View File

@@ -282,7 +282,7 @@ func (this *NSDomainDAO) NotifyUpdate(tx *dbs.Tx, domainId int64) error {
return err
}
if clusterId > 0 {
return models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, models.NSNodeTaskTypeDomainChanged)
return models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, 0, models.NSNodeTaskTypeDomainChanged)
}
return nil

2
internal/db/models/nameservers/ns_key_dao.go
View File

@@ -199,7 +199,7 @@ func (this *NSKeyDAO) NotifyUpdate(tx *dbs.Tx, keyId int64) error {
return err
}
if clusterId > 0 {
err = models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, models.NSNodeTaskTypeKeyChanged)
err = models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, 0, models.NSNodeTaskTypeKeyChanged)
if err != nil {
return err
}

2
internal/db/models/nameservers/ns_record_dao.go
View File

@@ -279,7 +279,7 @@ func (this *NSRecordDAO) NotifyUpdate(tx *dbs.Tx, recordId int64) error {
}
if clusterId > 0 {
err = models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, models.NSNodeTaskTypeRecordChanged)
err = models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, 0, models.NSNodeTaskTypeRecordChanged)
if err != nil {
return err
}

2
internal/db/models/nameservers/ns_route_dao.go
View File

@@ -259,7 +259,7 @@ func (this *NSRouteDAO) NotifyUpdate(tx *dbs.Tx) error {
return err
}
for _, clusterId := range clusterIds {
err = models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, models.NSNodeTaskTypeRouteChanged)
err = models.SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, 0, models.NSNodeTaskTypeRouteChanged)
if err != nil {
return err
}

2
internal/db/models/node_cluster_dao.go
View File

@@ -901,7 +901,7 @@ func (this *NodeClusterDAO) FindClusterBasicInfo(tx *dbs.Tx, clusterId int64, ca
// NotifyUpdate 通知更新
func (this *NodeClusterDAO) NotifyUpdate(tx *dbs.Tx, clusterId int64) error {
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeConfigChanged)
}
// NotifyDNSUpdate 通知DNS更新

2
internal/db/models/node_cluster_metric_item_dao.go
View File

@@ -177,5 +177,5 @@ func (this *NodeClusterMetricItemDAO) ExistsClusterItem(tx *dbs.Tx, clusterId in
// NotifyUpdate 通知更新
func (this *NodeClusterMetricItemDAO) NotifyUpdate(tx *dbs.Tx, clusterId int64) error {
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, 0, NodeTaskTypeConfigChanged)
}

2
internal/db/models/node_dao.go
View File

@@ -1471,7 +1471,7 @@ func (this *NodeDAO) NotifyUpdate(tx *dbs.Tx, nodeId int64) error {
return err
}
if clusterId > 0 {
return SharedNodeTaskDAO.CreateNodeTask(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, NodeTaskTypeConfigChanged, 0)
return SharedNodeTaskDAO.CreateNodeTask(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, 0, NodeTaskTypeConfigChanged, 0)
}
return nil
}

17
internal/db/models/node_task_dao.go
View File

@@ -49,7 +49,7 @@ func init() {
}
// CreateNodeTask 创建单个节点任务
func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, role string, clusterId int64, nodeId int64, taskType NodeTaskType, version int64) error {
func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, role string, clusterId int64, nodeId int64, serverId int64, taskType NodeTaskType, version int64) error {
if clusterId <= 0 || nodeId <= 0 {
return nil
}
@@ -60,6 +60,7 @@ func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, role string, clusterId int64
"role": role,
"clusterId": clusterId,
"nodeId": nodeId,
"serverId": serverId,
"type": taskType,
"uniqueId": uniqueId,
"updatedAt": updatedAt,
@@ -80,17 +81,18 @@ func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, role string, clusterId int64
}
// CreateClusterTask 创建集群任务
func (this *NodeTaskDAO) CreateClusterTask(tx *dbs.Tx, role string, clusterId int64, taskType NodeTaskType) error {
func (this *NodeTaskDAO) CreateClusterTask(tx *dbs.Tx, role string, clusterId int64, serverId int64, taskType NodeTaskType) error {
if clusterId <= 0 {
return nil
}
uniqueId := role + "@" + types.String(clusterId) + "@cluster@" + taskType
uniqueId := role + "@" + types.String(clusterId) + "@" + types.String(serverId) + "@cluster@" + taskType
updatedAt := time.Now().Unix()
_, _, err := this.Query(tx).
InsertOrUpdate(maps.Map{
"role": role,
"clusterId": clusterId,
"serverId": serverId,
"nodeId": 0,
"type": taskType,
"uniqueId": uniqueId,
@@ -112,7 +114,7 @@ func (this *NodeTaskDAO) CreateClusterTask(tx *dbs.Tx, role string, clusterId in
}
// ExtractNodeClusterTask 分解边缘节点集群任务
func (this *NodeTaskDAO) ExtractNodeClusterTask(tx *dbs.Tx, clusterId int64, taskType NodeTaskType) error {
func (this *NodeTaskDAO) ExtractNodeClusterTask(tx *dbs.Tx, clusterId int64, serverId int64, taskType NodeTaskType) error {
nodeIds, err := SharedNodeDAO.FindAllNodeIdsMatch(tx, clusterId, true, configutils.BoolStateYes)
if err != nil {
return err
@@ -131,7 +133,7 @@ func (this *NodeTaskDAO) ExtractNodeClusterTask(tx *dbs.Tx, clusterId int64, tas
var version = time.Now().UnixNano()
for _, nodeId := range nodeIds {
err = this.CreateNodeTask(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, taskType, version)
err = this.CreateNodeTask(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, serverId, taskType, version)
if err != nil {
return err
}
@@ -170,7 +172,7 @@ func (this *NodeTaskDAO) ExtractNSClusterTask(tx *dbs.Tx, clusterId int64, taskT
var version = time.Now().UnixNano()
for _, nodeId := range nodeIds {
err = this.CreateNodeTask(tx, nodeconfigs.NodeRoleDNS, clusterId, nodeId, taskType, version)
err = this.CreateNodeTask(tx, nodeconfigs.NodeRoleDNS, clusterId, nodeId, 0, taskType, version)
if err != nil {
return err
}
@@ -202,7 +204,8 @@ func (this *NodeTaskDAO) ExtractAllClusterTasks(tx *dbs.Tx, role string) error {
clusterId := int64(one.(*NodeTask).ClusterId)
switch role {
case nodeconfigs.NodeRoleNode:
err = this.ExtractNodeClusterTask(tx, clusterId, one.(*NodeTask).Type)
var nodeTask = one.(*NodeTask)
err = this.ExtractNodeClusterTask(tx, clusterId, int64(nodeTask.ServerId), nodeTask.Type)
if err != nil {
return err
}

2
internal/db/models/node_task_model.go
View File

@@ -6,6 +6,7 @@ type NodeTask struct {
Role string `field:"role"` // 节点角色
NodeId uint32 `field:"nodeId"` // 节点ID
ClusterId uint32 `field:"clusterId"` // 集群ID
ServerId uint32 `field:"serverId"` // 服务ID
Type string `field:"type"` // 任务类型
UniqueId string `field:"uniqueId"` // 唯一ID：nodeId@type
UpdatedAt uint64 `field:"updatedAt"` // 修改时间
@@ -21,6 +22,7 @@ type NodeTaskOperator struct {
Role interface{} // 节点角色
NodeId interface{} // 节点ID
ClusterId interface{} // 集群ID
ServerId interface{} // 服务ID
Type interface{} // 任务类型
UniqueId interface{} // 唯一ID：nodeId@type
UpdatedAt interface{} // 修改时间

2
internal/db/models/ns_cluster_dao.go
View File

@@ -195,5 +195,5 @@ func (this *NSClusterDAO) FindClusterRecursion(tx *dbs.Tx, clusterId int64) ([]b
// NotifyUpdate 通知更改
func (this *NSClusterDAO) NotifyUpdate(tx *dbs.Tx, clusterId int64) error {
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, NSNodeTaskTypeConfigChanged)
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, 0, NSNodeTaskTypeConfigChanged)
}

10
internal/db/models/server_dao.go
View File

@@ -1500,11 +1500,11 @@ func (this *ServerDAO) UpdateUserServersClusterId(tx *dbs.Tx, userId int64, oldC
}
if oldClusterId > 0 {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, oldClusterId, NodeTaskTypeConfigChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, oldClusterId, 0, NodeTaskTypeConfigChanged)
if err != nil {
return err
}
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, oldClusterId, NodeTaskTypeIPItemChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, oldClusterId, 0, NodeTaskTypeIPItemChanged)
if err != nil {
return err
}
@@ -1515,11 +1515,11 @@ func (this *ServerDAO) UpdateUserServersClusterId(tx *dbs.Tx, userId int64, oldC
}
if newClusterId > 0 {
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, newClusterId, NodeTaskTypeConfigChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, newClusterId, 0, NodeTaskTypeConfigChanged)
if err != nil {
return err
}
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, newClusterId, NodeTaskTypeIPItemChanged)
err = SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, newClusterId, 0, NodeTaskTypeIPItemChanged)
if err != nil {
return err
}
@@ -2260,7 +2260,7 @@ func (this *ServerDAO) NotifyUpdate(tx *dbs.Tx, serverId int64) error {
if clusterId == 0 {
return nil
}
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, serverId, NodeTaskTypeConfigChanged)
}
// NotifyDNSUpdate 通知DNS更新

3
internal/rpc/services/service_node_task.go
View File

@@ -39,6 +39,7 @@ func (this *NodeTaskService) FindNodeTasks(ctx context.Context, req *pb.FindNode
Type: task.Type,
Version: int64(task.Version),
IsPrimary: primaryNodeId == nodeId,
ServerId: int64(task.ServerId),
})
}
@@ -137,6 +138,7 @@ func (this *NodeTaskService) FindNodeClusterTasks(ctx context.Context, req *pb.F
IsOk: task.IsOk == 1,
Error: task.Error,
UpdatedAt: int64(task.UpdatedAt),
ServerId: int64(task.ServerId),
Node: &pb.Node{
Id: int64(task.NodeId),
Name: nodeName,
@@ -261,6 +263,7 @@ func (this *NodeTaskService) FindNotifyingNodeTasks(ctx context.Context, req *pb
Error: task.Error,
UpdatedAt: int64(task.UpdatedAt),
Node: &pb.Node{Id: int64(task.NodeId)},
ServerId: int64(task.ServerId),
})
}

26
internal/rpc/services/service_server.go
View File

@@ -1924,3 +1924,29 @@ func (this *ServerService) FindServerUserPlan(ctx context.Context, req *pb.FindS
},
}, nil
}
// ComposeServerConfig 获取服务配置
func (this *ServerService) ComposeServerConfig(ctx context.Context, req *pb.ComposeServerConfigRequest) (*pb.ComposeServerConfigResponse, error) {
_, err := this.ValidateNode(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
serverConfig, err := models.SharedServerDAO.ComposeServerConfigWithServerId(tx, req.ServerId, true)
if err != nil {
if err == models.ErrNotFound {
return &pb.ComposeServerConfigResponse{ServerConfigJSON: nil}, nil
}
return nil, err
}
if serverConfig == nil {
return &pb.ComposeServerConfigResponse{ServerConfigJSON: nil}, nil
}
configJSON, err := json.Marshal(serverConfig)
if err != nil {
return nil, err
}
return &pb.ComposeServerConfigResponse{ServerConfigJSON: configJSON}, nil
}