TRKJ/EdgeAPI
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeAPI.git synced 2025-12-15 07:06:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

数据有更改时发送通知

Browse Source
This commit is contained in:
GoEdgeLab
2021-08-08 15:47:48 +08:00
parent 07aa0990f8
commit e979414973
40 changed files with 548 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,8 +1,8 @@
package models
import (
"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
_ "github.com/go-sql-driver/mysql"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/dbs"
@@ -17,6 +17,14 @@ const (
NodeTaskTypeConfigChanged NodeTaskType = "configChanged"
NodeTaskTypeIPItemChanged NodeTaskType = "ipItemChanged"
NodeTaskTypeNodeVersionChanged NodeTaskType = "nodeVersionChanged"
// NS相关
NSNodeTaskTypeConfigChanged NodeTaskType = "nsConfigChanged"
NSNodeTaskTypeDomainChanged NodeTaskType = "nsDomainChanged"
NSNodeTaskTypeRecordChanged NodeTaskType = "nsRecordChanged"
NSNodeTaskTypeRouteChanged NodeTaskType = "nsRouteChanged"
NSNodeTaskTypeKeyChanged NodeTaskType = "nsKeyChanged"
)
type NodeTaskDAO dbs.DAO
@@ -41,14 +49,15 @@ func init() {
}
// CreateNodeTask 创建单个节点任务
func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, clusterId int64, nodeId int64, taskType NodeTaskType) error {
func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, role string, clusterId int64, nodeId int64, taskType NodeTaskType) error {
if clusterId <= 0 || nodeId <= 0 {
return nil
}
uniqueId := numberutils.FormatInt64(nodeId) + "@node@" + taskType
uniqueId := role + "@" + types.String(nodeId) + "@node@" + taskType
updatedAt := time.Now().Unix()
_, _, err := this.Query(tx).
InsertOrUpdate(maps.Map{
"role": role,
"clusterId": clusterId,
"nodeId": nodeId,
"type": taskType,
@@ -68,15 +77,16 @@ func (this *NodeTaskDAO) CreateNodeTask(tx *dbs.Tx, clusterId int64, nodeId int6
}
// CreateClusterTask 创建集群任务
func (this *NodeTaskDAO) CreateClusterTask(tx *dbs.Tx, clusterId int64, taskType NodeTaskType) error {
func (this *NodeTaskDAO) CreateClusterTask(tx *dbs.Tx, role string, clusterId int64, taskType NodeTaskType) error {
if clusterId <= 0 {
return nil
}
uniqueId := numberutils.FormatInt64(clusterId) + "@cluster@" + taskType
uniqueId := role + "@" + types.String(clusterId) + "@cluster@" + taskType
updatedAt := time.Now().Unix()
_, _, err := this.Query(tx).
InsertOrUpdate(maps.Map{
"role": role,
"clusterId": clusterId,
"nodeId": 0,
"type": taskType,
@@ -96,14 +106,15 @@ func (this *NodeTaskDAO) CreateClusterTask(tx *dbs.Tx, clusterId int64, taskType
return err
}
// ExtractClusterTask 分解集群任务
func (this *NodeTaskDAO) ExtractClusterTask(tx *dbs.Tx, clusterId int64, taskType NodeTaskType) error {
// ExtractNodeClusterTask 分解边缘节点集群任务
func (this *NodeTaskDAO) ExtractNodeClusterTask(tx *dbs.Tx, clusterId int64, taskType NodeTaskType) error {
nodeIds, err := SharedNodeDAO.FindAllNodeIdsMatch(tx, clusterId, true, configutils.BoolStateYes)
if err != nil {
return err
}
_, err = this.Query(tx).
Attr("role", nodeconfigs.NodeRoleNode).
Attr("clusterId", clusterId).
Param("clusterIdString", types.String(clusterId)).
Where("nodeId> 0").
@@ -114,13 +125,52 @@ func (this *NodeTaskDAO) ExtractClusterTask(tx *dbs.Tx, clusterId int64, taskTyp
}
for _, nodeId := range nodeIds {
err = this.CreateNodeTask(tx, clusterId, nodeId, taskType)
err = this.CreateNodeTask(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, taskType)
if err != nil {
return err
}
}
_, err = this.Query(tx).
Attr("role", nodeconfigs.NodeRoleNode).
Attr("clusterId", clusterId).
Attr("nodeId", 0).
Attr("type", taskType).
Delete()
if err != nil {
return err
}
return nil
}
// ExtractNSClusterTask 分解NS节点集群任务
func (this *NodeTaskDAO) ExtractNSClusterTask(tx *dbs.Tx, clusterId int64, taskType NodeTaskType) error {
nodeIds, err := SharedNSNodeDAO.FindAllNodeIdsMatch(tx, clusterId, true, configutils.BoolStateYes)
if err != nil {
return err
}
_, err = this.Query(tx).
Attr("role", nodeconfigs.NodeRoleDNS).
Attr("clusterId", clusterId).
Param("clusterIdString", types.String(clusterId)).
Where("nodeId> 0").
Attr("type", taskType).
Delete()
if err != nil {
return err
}
for _, nodeId := range nodeIds {
err = this.CreateNodeTask(tx, nodeconfigs.NodeRoleDNS, clusterId, nodeId, taskType)
if err != nil {
return err
}
}
_, err = this.Query(tx).
Attr("role", nodeconfigs.NodeRoleDNS).
Attr("clusterId", clusterId).
Attr("nodeId", 0).
Attr("type", taskType).
@@ -133,8 +183,9 @@ func (this *NodeTaskDAO) ExtractClusterTask(tx *dbs.Tx, clusterId int64, taskTyp
}
// ExtractAllClusterTasks 分解所有集群任务
func (this *NodeTaskDAO) ExtractAllClusterTasks(tx *dbs.Tx) error {
func (this *NodeTaskDAO) ExtractAllClusterTasks(tx *dbs.Tx, role string) error {
ones, err := this.Query(tx).
Attr("role", role).
Attr("nodeId", 0).
FindAll()
if err != nil {
@@ -142,36 +193,47 @@ func (this *NodeTaskDAO) ExtractAllClusterTasks(tx *dbs.Tx) error {
}
for _, one := range ones {
clusterId := int64(one.(*NodeTask).ClusterId)
err = this.ExtractClusterTask(tx, clusterId, one.(*NodeTask).Type)
if err != nil {
return err
switch role {
case nodeconfigs.NodeRoleNode:
err = this.ExtractNodeClusterTask(tx, clusterId, one.(*NodeTask).Type)
if err != nil {
return err
}
case nodeconfigs.NodeRoleDNS:
err = this.ExtractNSClusterTask(tx, clusterId, one.(*NodeTask).Type)
if err != nil {
return err
}
}
}
return nil
}
// DeleteAllClusterTasks 删除集群所有相关任务
func (this *NodeTaskDAO) DeleteAllClusterTasks(tx *dbs.Tx, clusterId int64) error {
func (this *NodeTaskDAO) DeleteAllClusterTasks(tx *dbs.Tx, role string, clusterId int64) error {
_, err := this.Query(tx).
Attr("role", role).
Attr("clusterId", clusterId).
Delete()
return err
}
// DeleteNodeTasks 删除节点相关任务
func (this *NodeTaskDAO) DeleteNodeTasks(tx *dbs.Tx, nodeId int64) error {
func (this *NodeTaskDAO) DeleteNodeTasks(tx *dbs.Tx, role string, nodeId int64) error {
_, err := this.Query(tx).
Attr("role", role).
Attr("nodeId", nodeId).
Delete()
return err
}
// FindDoingNodeTasks 查询一个节点的所有任务
func (this *NodeTaskDAO) FindDoingNodeTasks(tx *dbs.Tx, nodeId int64) (result []*NodeTask, err error) {
func (this *NodeTaskDAO) FindDoingNodeTasks(tx *dbs.Tx, role string, nodeId int64) (result []*NodeTask, err error) {
if nodeId <= 0 {
return
}
_, err = this.Query(tx).
Attr("role", role).
Attr("nodeId", nodeId).
Where("(isDone=0 OR (isDone=1 AND isOk=0))").
Slice(&result).
@@ -191,9 +253,10 @@ func (this *NodeTaskDAO) UpdateNodeTaskDone(tx *dbs.Tx, taskId int64, isOk bool,
}
// FindAllDoingTaskClusterIds 查找正在更新的集群IDs
func (this *NodeTaskDAO) FindAllDoingTaskClusterIds(tx *dbs.Tx) ([]int64, error) {
func (this *NodeTaskDAO) FindAllDoingTaskClusterIds(tx *dbs.Tx, role string) ([]int64, error) {
ones, _, err := this.Query(tx).
Result("DISTINCT(clusterId) AS clusterId").
Attr("role", role).
Where("(nodeId=0 OR (isDone=0 OR (isDone=1 AND isOk=0)))").
FindOnes()
if err != nil {
@@ -207,8 +270,9 @@ func (this *NodeTaskDAO) FindAllDoingTaskClusterIds(tx *dbs.Tx) ([]int64, error)
}
// FindAllDoingNodeTasksWithClusterId 查询某个集群下所有的任务
func (this *NodeTaskDAO) FindAllDoingNodeTasksWithClusterId(tx *dbs.Tx, clusterId int64) (result []*NodeTask, err error) {
func (this *NodeTaskDAO) FindAllDoingNodeTasksWithClusterId(tx *dbs.Tx, role string, clusterId int64) (result []*NodeTask, err error) {
_, err = this.Query(tx).
Attr("role", role).
Attr("clusterId", clusterId).
Gt("nodeId", 0).
Where("(isDone=0 OR (isDone=1 AND isOk=0))").
@@ -220,17 +284,37 @@ func (this *NodeTaskDAO) FindAllDoingNodeTasksWithClusterId(tx *dbs.Tx, clusterI
return
}
// FindAllDoingNodeIds 查询有任务的节点IDs
func (this *NodeTaskDAO) FindAllDoingNodeIds(tx *dbs.Tx, role string) ([]int64, error) {
ones, err := this.Query(tx).
Result("DISTINCT(nodeId) AS nodeId").
Attr("role", role).
Gt("nodeId", 0).
Attr("isDone", false).
FindAll()
if err != nil {
return nil, err
}
var result []int64
for _, one := range ones {
result = append(result, int64(one.(*NodeTask).NodeId))
}
return result, nil
}
// ExistsDoingNodeTasks 检查是否有正在执行的任务
func (this *NodeTaskDAO) ExistsDoingNodeTasks(tx *dbs.Tx) (bool, error) {
func (this *NodeTaskDAO) ExistsDoingNodeTasks(tx *dbs.Tx, role string) (bool, error) {
return this.Query(tx).
Attr("role", role).
Where("(isDone=0 OR (isDone=1 AND isOk=0))").
Gt("nodeId", 0).
Exist()
}
// ExistsErrorNodeTasks 是否有错误的任务
func (this *NodeTaskDAO) ExistsErrorNodeTasks(tx *dbs.Tx) (bool, error) {
func (this *NodeTaskDAO) ExistsErrorNodeTasks(tx *dbs.Tx, role string) (bool, error) {
return this.Query(tx).
Attr("role", role).
Where("(isDone=1 AND isOk=0)").
Exist()
}
@@ -244,16 +328,18 @@ func (this *NodeTaskDAO) DeleteNodeTask(tx *dbs.Tx, taskId int64) error {
}
// CountDoingNodeTasks 计算正在执行的任务
func (this *NodeTaskDAO) CountDoingNodeTasks(tx *dbs.Tx) (int64, error) {
func (this *NodeTaskDAO) CountDoingNodeTasks(tx *dbs.Tx, role string) (int64, error) {
return this.Query(tx).
Attr("isDone", 0).
Attr("role", role).
Gt("nodeId", 0).
Count()
}
// FindNotifyingNodeTasks 查找需要通知的任务
func (this *NodeTaskDAO) FindNotifyingNodeTasks(tx *dbs.Tx, size int64) (result []*NodeTask, err error) {
func (this *NodeTaskDAO) FindNotifyingNodeTasks(tx *dbs.Tx, role string, size int64) (result []*NodeTask, err error) {
_, err = this.Query(tx).
Attr("role", role).
Gt("nodeId", 0).
Attr("isNotified", 0).
Attr("isDone", 0).