✨ [dashboard v0.3.9] 手动即时触发计划任务

2026-02-05 21:20:06 +00:00 · 2021-01-24 09:41:35 +08:00
parent 175eb996d3
commit a957b21265
10 changed files with 195 additions and 116 deletions
--- a/service/alertmanager/alertmanager.go
+++ b/service/alertmanager/alertmanager.go
@@ -1,23 +1,14 @@
-package alertmanager
+package dao

 import (
-	"crypto/md5"
-	"encoding/hex"
 	"fmt"
 	"log"
 	"sync"
 	"time"

 	"github.com/naiba/nezha/model"
-	"github.com/naiba/nezha/service/dao"
 )

-const firstNotificationDelay = time.Minute * 15
-
-// 通知方式
-var notifications []model.Notification
-var notificationsLock sync.RWMutex
-
 // 报警规则
 var alertsLock sync.RWMutex
 var alerts []model.AlertRule
@@ -28,15 +19,15 @@ type NotificationHistory struct {
 	Until    time.Time
 }

-func Start() {
+func AlertSentinelStart() {
 	alertsStore = make(map[uint64]map[uint64][][]interface{})
 	notificationsLock.Lock()
-	if err := dao.DB.Find(&notifications).Error; err != nil {
+	if err := DB.Find(&notifications).Error; err != nil {
 		panic(err)
 	}
 	notificationsLock.Unlock()
 	alertsLock.Lock()
-	if err := dao.DB.Find(&alerts).Error; err != nil {
+	if err := DB.Find(&alerts).Error; err != nil {
 		panic(err)
 	}
 	for i := 0; i < len(alerts); i++ {
@@ -56,7 +47,7 @@ func Start() {
 			checkCount = 0
 			lastPrint = startedAt
 		}
-		time.Sleep(time.Until(startedAt.Add(time.Second * dao.SnapshotDelay)))
+		time.Sleep(time.Until(startedAt.Add(time.Second * SnapshotDelay)))
 	}
 }

@@ -89,44 +80,18 @@ func OnDeleteAlert(id uint64) {
 	}
 }

-func OnRefreshOrAddNotification(n model.Notification) {
-	notificationsLock.Lock()
-	defer notificationsLock.Unlock()
-	var isEdit bool
-	for i := 0; i < len(notifications); i++ {
-		if notifications[i].ID == n.ID {
-			notifications[i] = n
-			isEdit = true
-		}
-	}
-	if !isEdit {
-		notifications = append(notifications, n)
-	}
-}
-
-func OnDeleteNotification(id uint64) {
-	notificationsLock.Lock()
-	defer notificationsLock.Unlock()
-	for i := 0; i < len(notifications); i++ {
-		if notifications[i].ID == id {
-			notifications = append(notifications[:i], notifications[i+1:]...)
-			i--
-		}
-	}
-}
-
 func checkStatus() {
 	alertsLock.RLock()
 	defer alertsLock.RUnlock()
-	dao.ServerLock.RLock()
-	defer dao.ServerLock.RUnlock()
+	ServerLock.RLock()
+	defer ServerLock.RUnlock()

 	for _, alert := range alerts {
 		// 跳过未启用
 		if alert.Enable == nil || !*alert.Enable {
 			continue
 		}
-		for _, server := range dao.ServerList {
+		for _, server := range ServerList {
 			// 监测点
 			alertsStore[alert.ID][server.ID] = append(alertsStore[alert.
 				ID][server.ID], alert.Snapshot(server))
@@ -143,42 +108,3 @@ func checkStatus() {
 		}
 	}
 }
-
-func SendNotification(desc string, muteable bool) {
-	if muteable {
-		// 通知防骚扰策略
-		nID := hex.EncodeToString(md5.New().Sum([]byte(desc)))
-		var flag bool
-		if cacheN, has := dao.Cache.Get(nID); has {
-			nHistory := cacheN.(NotificationHistory)
-			// 每次提醒都增加一倍等待时间，最后每天最多提醒一次
-			if time.Now().After(nHistory.Until) {
-				flag = true
-				nHistory.Duration *= 2
-				if nHistory.Duration > time.Hour*24 {
-					nHistory.Duration = time.Hour * 24
-				}
-				nHistory.Until = time.Now().Add(nHistory.Duration)
-				// 缓存有效期加 10 分钟
-				dao.Cache.Set(nID, nHistory, nHistory.Duration+time.Minute*10)
-			}
-		} else {
-			// 新提醒直接通知
-			flag = true
-			dao.Cache.Set(nID, NotificationHistory{
-				Duration: firstNotificationDelay,
-				Until:    time.Now().Add(firstNotificationDelay),
-			}, firstNotificationDelay+time.Minute*10)
-		}
-
-		if !flag {
-			return
-		}
-	}
-	// 发出通知
-	notificationsLock.RLock()
-	defer notificationsLock.RUnlock()
-	for i := 0; i < len(notifications); i++ {
-		notifications[i].Send(desc)
-	}
-}
--- a/service/dao/dao.go
+++ b/service/dao/dao.go
@@ -1,6 +1,7 @@
 package dao

 import (
+	"fmt"
 	"sort"
 	"sync"

@@ -9,32 +10,27 @@ import (
 	"gorm.io/gorm"

 	"github.com/naiba/nezha/model"
+	pb "github.com/naiba/nezha/proto"
 )

+var Version = "v0.3.9"
+
 const (
 	SnapshotDelay = 3
 	ReportDelay   = 2
 )

-var Conf *model.Config
+var (
+	Conf  *model.Config
+	Cache *cache.Cache
+	DB    *gorm.DB

-var Cache *cache.Cache
+	ServerList map[uint64]*model.Server
+	ServerLock sync.RWMutex

-var DB *gorm.DB
-
-// 服务器监控、状态相关
-var ServerList map[uint64]*model.Server
-var ServerLock sync.RWMutex
-
-var SortedServerList []*model.Server
-var SortedServerLock sync.RWMutex
-
-// 计划任务相关
-var CronLock sync.RWMutex
-var Crons map[uint64]*model.Cron
-var Cron *cron.Cron
-
-var Version = "v0.3.8"
+	SortedServerList []*model.Server
+	SortedServerLock sync.RWMutex
+)

 func ReSortServer() {
 	ServerLock.RLock()
@@ -54,3 +50,25 @@ func ReSortServer() {
 		return SortedServerList[i].DisplayIndex > SortedServerList[j].DisplayIndex
 	})
 }
+
+// =============== Cron Mixin ===============
+
+var CronLock sync.RWMutex
+var Crons map[uint64]*model.Cron
+var Cron *cron.Cron
+
+func CronTrigger(c *model.Cron) {
+	ServerLock.RLock()
+	defer ServerLock.RUnlock()
+	for j := 0; j < len(c.Servers); j++ {
+		if ServerList[c.Servers[j]].TaskStream != nil {
+			ServerList[c.Servers[j]].TaskStream.Send(&pb.Task{
+				Id:   c.ID,
+				Data: c.Command,
+				Type: model.TaskTypeCommand,
+			})
+		} else {
+			SendNotification(fmt.Sprintf("计划任务：%s，服务器：%d 离线，无法执行。", c.Name, c.Servers[j]), false)
+		}
+	}
+}
--- a/service/dao/notification.go
+++ b/service/dao/notification.go
@@ -0,0 +1,81 @@
+package dao
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"sync"
+	"time"
+
+	"github.com/naiba/nezha/model"
+)
+
+const firstNotificationDelay = time.Minute * 15
+
+// 通知方式
+var notifications []model.Notification
+var notificationsLock sync.RWMutex
+
+func OnRefreshOrAddNotification(n model.Notification) {
+	notificationsLock.Lock()
+	defer notificationsLock.Unlock()
+	var isEdit bool
+	for i := 0; i < len(notifications); i++ {
+		if notifications[i].ID == n.ID {
+			notifications[i] = n
+			isEdit = true
+		}
+	}
+	if !isEdit {
+		notifications = append(notifications, n)
+	}
+}
+
+func OnDeleteNotification(id uint64) {
+	notificationsLock.Lock()
+	defer notificationsLock.Unlock()
+	for i := 0; i < len(notifications); i++ {
+		if notifications[i].ID == id {
+			notifications = append(notifications[:i], notifications[i+1:]...)
+			i--
+		}
+	}
+}
+
+func SendNotification(desc string, muteable bool) {
+	if muteable {
+		// 通知防骚扰策略
+		nID := hex.EncodeToString(md5.New().Sum([]byte(desc)))
+		var flag bool
+		if cacheN, has := Cache.Get(nID); has {
+			nHistory := cacheN.(NotificationHistory)
+			// 每次提醒都增加一倍等待时间，最后每天最多提醒一次
+			if time.Now().After(nHistory.Until) {
+				flag = true
+				nHistory.Duration *= 2
+				if nHistory.Duration > time.Hour*24 {
+					nHistory.Duration = time.Hour * 24
+				}
+				nHistory.Until = time.Now().Add(nHistory.Duration)
+				// 缓存有效期加 10 分钟
+				Cache.Set(nID, nHistory, nHistory.Duration+time.Minute*10)
+			}
+		} else {
+			// 新提醒直接通知
+			flag = true
+			Cache.Set(nID, NotificationHistory{
+				Duration: firstNotificationDelay,
+				Until:    time.Now().Add(firstNotificationDelay),
+			}, firstNotificationDelay+time.Minute*10)
+		}
+
+		if !flag {
+			return
+		}
+	}
+	// 发出通知
+	notificationsLock.RLock()
+	defer notificationsLock.RUnlock()
+	for i := 0; i < len(notifications); i++ {
+		notifications[i].Send(desc)
+	}
+}
--- a/service/rpc/nezha.go
+++ b/service/rpc/nezha.go
@@ -8,7 +8,6 @@ import (

 	"github.com/naiba/nezha/model"
 	pb "github.com/naiba/nezha/proto"
-	"github.com/naiba/nezha/service/alertmanager"
 	"github.com/naiba/nezha/service/dao"
 )

@@ -52,7 +51,7 @@ func (s *NezhaHandler) ReportTask(c context.Context, r *pb.TaskResult) (*pb.Rece
 		if errMsg != "" {
 			var monitor model.Monitor
 			dao.DB.First(&monitor, "id = ?", r.GetId())
-			alertmanager.SendNotification(fmt.Sprintf("服务监控：%s %s", monitor.Name, errMsg), true)
+			dao.SendNotification(fmt.Sprintf("服务监控：%s %s", monitor.Name, errMsg), true)
 		}
 	}
 	if r.GetType() == model.TaskTypeCommand {
@@ -62,10 +61,10 @@ func (s *NezhaHandler) ReportTask(c context.Context, r *pb.TaskResult) (*pb.Rece
 		cr := dao.Crons[r.GetId()]
 		if cr != nil {
 			if cr.PushSuccessful && r.GetSuccessful() {
-				alertmanager.SendNotification(fmt.Sprintf("成功计划任务：%s ，服务器：%d，日志：\n%s", cr.Name, clientID, r.GetData()), false)
+				dao.SendNotification(fmt.Sprintf("成功计划任务：%s ，服务器：%d，日志：\n%s", cr.Name, clientID, r.GetData()), false)
 			}
 			if !r.GetSuccessful() {
-				alertmanager.SendNotification(fmt.Sprintf("失败计划任务：%s ，服务器：%d，日志：\n%s", cr.Name, clientID, r.GetData()), false)
+				dao.SendNotification(fmt.Sprintf("失败计划任务：%s ，服务器：%d，日志：\n%s", cr.Name, clientID, r.GetData()), false)
 			}
 			dao.DB.Model(cr).Updates(model.Cron{
 				LastExecutedAt: time.Now().Add(time.Second * -1 * time.Duration(r.GetDelay())),
@@ -127,7 +126,7 @@ func (s *NezhaHandler) ReportSystemInfo(c context.Context, r *pb.Host) (*pb.Rece
 		dao.ServerList[clientID].Host.IP != "" &&
 		host.IP != "" &&
 		dao.ServerList[clientID].Host.IP != host.IP {
-		alertmanager.SendNotification(fmt.Sprintf(
+		dao.SendNotification(fmt.Sprintf(
 			"IP变更提醒 服务器：%s ，旧IP：%s，新IP：%s。",
 			dao.ServerList[clientID].Name, dao.ServerList[clientID].Host.IP, host.IP), true)
 	}