增加健康检查定时任务/健康检查可以发送消息

internal/tasks/event_looper.go

@@ -2,13 +2,16 @@ package tasks
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/logs"
 	"time"
 )
 
 func init() {
-	looper := NewEventLooper()
-	go looper.Start()
+	dbs.OnReady(func() {
+		looper := NewEventLooper()
+		go looper.Start()
+	})
 }
 
 type EventLooper struct {

internal/tasks/health_check_cluster_task.go

@@ -0,0 +1,145 @@
+package tasks
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
+	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/iwind/TeaGo/logs"
+	"github.com/iwind/TeaGo/maps"
+	"time"
+)
+
+// 单个集群的健康检查任务
+type HealthCheckClusterTask struct {
+	clusterId int64
+	config    *serverconfigs.HealthCheckConfig
+	ticker    *utils.Ticker
+}
+
+// 创建新任务
+func NewHealthCheckClusterTask(clusterId int64, config *serverconfigs.HealthCheckConfig) *HealthCheckClusterTask {
+	return &HealthCheckClusterTask{
+		clusterId: clusterId,
+		config:    config,
+	}
+}
+
+// 重置配置
+func (this *HealthCheckClusterTask) Reset(config *serverconfigs.HealthCheckConfig) {
+	// 检查是否有变化
+	oldJSON, err := json.Marshal(this.config)
+	if err != nil {
+		logs.Println("[TASK][HEALTH_CHECK]" + err.Error())
+		return
+	}
+	newJSON, err := json.Marshal(config)
+	if err != nil {
+		logs.Println("[TASK][HEALTH_CHECK]" + err.Error())
+		return
+	}
+	if bytes.Compare(oldJSON, newJSON) != 0 {
+		this.config = config
+		this.Run()
+	}
+}
+
+// 执行
+func (this *HealthCheckClusterTask) Run() {
+	this.Stop()
+
+	if this.config == nil {
+		return
+	}
+	if !this.config.IsOn {
+		return
+	}
+	if this.config.Interval == nil {
+		return
+	}
+	duration := this.config.Interval.Duration()
+	if duration <= 0 {
+		return
+	}
+	ticker := utils.NewTicker(duration)
+	go func() {
+		for ticker.Wait() {
+			err := this.loop(int64(duration.Seconds()))
+			if err != nil {
+				logs.Println("[TASK][HEALTH_CHECK]" + err.Error())
+			}
+		}
+	}()
+	this.ticker = ticker
+}
+
+// 停止
+func (this *HealthCheckClusterTask) Stop() {
+	if this.ticker == nil {
+		return
+	}
+	this.ticker.Stop()
+	this.ticker = nil
+}
+
+// 单个循环任务
+func (this *HealthCheckClusterTask) loop(seconds int64) error {
+	// 检查上次运行时间，防止重复运行
+	settingKey := "cluster_health_check_%d" + numberutils.FormatInt64(this.clusterId)
+	timestamp := time.Now().Unix()
+	c, err := models.SharedSysSettingDAO.CompareInt64Setting(settingKey, timestamp-seconds)
+	if err != nil {
+		return err
+	}
+	if c > 0 {
+		return nil
+	}
+
+	// 记录时间
+	err = models.SharedSysSettingDAO.UpdateSetting(settingKey, []byte(numberutils.FormatInt64(timestamp)))
+	if err != nil {
+		return err
+	}
+
+	// 开始运行
+	executor := NewHealthCheckExecutor(this.clusterId)
+	results, err := executor.Run()
+	if err != nil {
+		return err
+	}
+
+	failedResults := []maps.Map{}
+	for _, result := range results {
+		if !result.IsOk {
+			failedResults = append(failedResults, maps.Map{
+				"node": maps.Map{
+					"id":   result.Node.Id,
+					"name": result.Node.Name,
+				},
+				"isOk":     false,
+				"error":    result.Error,
+				"nodeAddr": result.NodeAddr,
+			})
+		}
+	}
+
+	if len(failedResults) > 0 {
+		failedResultsJSON, err := json.Marshal(failedResults)
+		if err != nil {
+			return err
+		}
+		err = models.NewMessageDAO().CreateClusterMessage(this.clusterId, models.MessageTypeHealthCheckFail, models.MessageLevelError, "有"+numberutils.FormatInt(len(failedResults))+"个节点在健康检查中出现问题", failedResultsJSON)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// 获取当前配置
+func (this *HealthCheckClusterTask) Config() *serverconfigs.HealthCheckConfig {
+	return this.config
+}

internal/tasks/health_check_cluster_task_test.go

@@ -0,0 +1,16 @@
+package tasks
+
+import (
+	"github.com/iwind/TeaGo/dbs"
+	"testing"
+)
+
+func TestHealthCheckClusterTask_loop(t *testing.T) {
+	dbs.NotifyReady()
+	task := NewHealthCheckClusterTask(10, nil)
+	err := task.loop(10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok")
+}

internal/tasks/health_check_executor.go

@@ -24,7 +24,7 @@ func NewHealthCheckExecutor(clusterId int64) *HealthCheckExecutor {
 }
 
 func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
-	cluster, err := models.SharedNodeClusterDAO.FindEnabledNodeCluster(this.clusterId)
+	cluster, err := models.NewNodeClusterDAO().FindEnabledNodeCluster(this.clusterId)
 	if err != nil {
 		return nil, err
 	}
@@ -42,7 +42,7 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
 	}
 
 	results := []*HealthCheckResult{}
-	nodes, err := models.SharedNodeDAO.FindAllEnabledNodesWithClusterId(this.clusterId)
+	nodes, err := models.NewNodeDAO().FindAllEnabledNodesWithClusterId(this.clusterId)
 	if err != nil {
 		return nil, err
 	}
@@ -54,7 +54,7 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
 			Node: node,
 		}
 
-		addresses, err := models.SharedNodeIPAddressDAO.FindAllEnabledAddressesWithNode(int64(node.Id))
+		addresses, err := models.NewNodeIPAddressDAO().FindAllEnabledAddressesWithNode(int64(node.Id))
 		if err != nil {
 			return nil, err
 		}

internal/tasks/health_check_task.go

@@ -0,0 +1,95 @@
+package tasks
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
+	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/lists"
+	"github.com/iwind/TeaGo/logs"
+	"time"
+)
+
+func init() {
+	dbs.OnReady(func() {
+		go NewHealthCheckTask().Run()
+	})
+}
+
+type HealthCheckTask struct {
+	tasksMap map[int64]*HealthCheckClusterTask // taskId => task
+}
+
+func NewHealthCheckTask() *HealthCheckTask {
+	return &HealthCheckTask{
+		tasksMap: map[int64]*HealthCheckClusterTask{},
+	}
+}
+
+func (this *HealthCheckTask) Run() {
+	err := this.loop()
+	if err != nil {
+		logs.Println("[TASK][HEALTH_CHECK]" + err.Error())
+	}
+
+	ticker := utils.NewTicker(60 * time.Second)
+	for ticker.Wait() {
+		err := this.loop()
+		if err != nil {
+			logs.Println("[TASK][HEALTH_CHECK]" + err.Error())
+		}
+	}
+}
+
+func (this *HealthCheckTask) loop() error {
+	clusters, err := models.NewNodeClusterDAO().FindAllEnableClusters()
+	if err != nil {
+		return err
+	}
+	clusterIds := []int64{}
+	for _, cluster := range clusters {
+		clusterIds = append(clusterIds, int64(cluster.Id))
+	}
+
+	// 停掉删除的
+	for clusterId, task := range this.tasksMap {
+		if !lists.ContainsInt64(clusterIds, clusterId) {
+			task.Stop()
+			delete(this.tasksMap, clusterId)
+		}
+	}
+
+	// 启动新的或更新老的
+	for _, cluster := range clusters {
+		clusterId := int64(cluster.Id)
+
+		config := &serverconfigs.HealthCheckConfig{}
+		if len(cluster.HealthCheck) > 0 && cluster.HealthCheck != "null" {
+			err = json.Unmarshal([]byte(cluster.HealthCheck), config)
+			if err != nil {
+				logs.Println("[TASK][HEALTH_CHECK]" + err.Error())
+				continue
+			}
+		}
+
+		task, ok := this.tasksMap[clusterId]
+		if ok {
+			// 检查是否有变化
+			newJSON, _ := json.Marshal(config)
+			oldJSON, _ := json.Marshal(task.Config())
+			if bytes.Compare(oldJSON, newJSON) != 0 {
+				logs.Println("[TASK][HEALTH_CHECK]update cluster '" + numberutils.FormatInt64(clusterId) + "'")
+				go task.Reset(config)
+			}
+		} else {
+			task := NewHealthCheckClusterTask(clusterId, config)
+			this.tasksMap[clusterId] = task
+			go task.Run()
+		}
+	}
+
+	return nil
+}

internal/tasks/message_task.go

@@ -0,0 +1,41 @@
+package tasks
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/logs"
+	"time"
+)
+
+func init() {
+	dbs.OnReady(func() {
+		go NewMessageTask().Run()
+	})
+}
+
+// 消息相关任务
+type MessageTask struct {
+}
+
+// 获取新对象
+func NewMessageTask() *MessageTask {
+	return &MessageTask{}
+}
+
+// 运行
+func (this *MessageTask) Run() {
+	ticker := utils.NewTicker(24 * time.Hour)
+	for ticker.Wait() {
+		err := this.loop()
+		if err != nil {
+			logs.Println("[TASK][MESSAGE]" + err.Error())
+		}
+	}
+}
+
+// 单次运行
+func (this *MessageTask) loop() error {
+	dayTime := time.Now().AddDate(0, 0, -30) // TODO 这个30天应该可以在界面上设置
+	return models.NewMessageDAO().DeleteMessagesBeforeDay(dayTime)
+}

internal/tasks/node_log_cleaner_task.go

@@ -2,26 +2,29 @@ package tasks
 
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/logs"
 	"time"
 )
 
 func init() {
-	go NewNodeLogCleaner().Start()
+	dbs.OnReady(func() {
+		go NewNodeLogCleanerTask().Start()
+	})
 }
 
 // 清理节点日志的工具
-type NodeLogCleaner struct {
+type NodeLogCleanerTask struct {
 	duration time.Duration
 }
 
-func NewNodeLogCleaner() *NodeLogCleaner {
-	return &NodeLogCleaner{
+func NewNodeLogCleanerTask() *NodeLogCleanerTask {
+	return &NodeLogCleanerTask{
 		duration: 24 * time.Hour,
 	}
 }
 
-func (this *NodeLogCleaner) Start() {
+func (this *NodeLogCleanerTask) Start() {
 	ticker := time.NewTicker(this.duration)
 	for range ticker.C {
 		err := this.loop()
@@ -31,7 +34,7 @@ func (this *NodeLogCleaner) Start() {
 	}
 }
 
-func (this *NodeLogCleaner) loop() error {
+func (this *NodeLogCleanerTask) loop() error {
 	// TODO 30天这个数值改成可以设置
 	return models.SharedNodeLogDAO.DeleteExpiredLogs(30)
 }

internal/tasks/node_log_cleaner_task_test.go

@@ -3,7 +3,7 @@ package tasks
 import "testing"
 
 func TestNodeLogCleaner_loop(t *testing.T) {
-	cleaner := &NodeLogCleaner{}
+	cleaner := &NodeLogCleanerTask{}
 	err := cleaner.loop()
 	if err != nil {
 		t.Fatal(err)