feature: load alert_events to memory when start

alert/alert.go

@@ -1,6 +1,8 @@
 package alert
 
-import "context"
+import (
+	"context"
+)
 
 func Start(ctx context.Context) {
 	go popEvent()

judge/judge.go

@@ -28,6 +28,9 @@ func Start(ctx context.Context) {
 	// PUSH型的告警引擎，依赖内存里缓存的数据来做告警判断，两层map减小锁粒度
 	initPointCaches()
 
+	// 把数据库中的未恢复告警同步一份到内存中，便于后续判断告警是否应该发送
+	LastEvents.Init()
+
 	// 默认初始化的大小是1000万，相当于内存里有1000万事件，应该够用了
 	EventQueue = list.NewSafeListLimited(10000000)
 

judge/last_event.go

@@ -1,6 +1,8 @@
 package judge
 
 import (
+	"fmt"
+	"os"
 	"sync"
 	"time"
 
@@ -35,9 +37,9 @@ func (s *SafeEventMap) Set(event *models.AlertEvent) {
 	s.Lock()
 	defer s.Unlock()
 
-	m, has := s.M[event.RuleId]
+	_, has := s.M[event.RuleId]
 	if !has {
-		m = make(map[string]*models.AlertEvent)
+		m := make(map[string]*models.AlertEvent)
 		m[event.HashId] = event
 		s.M[event.RuleId] = m
 	} else {
@@ -45,6 +47,35 @@ func (s *SafeEventMap) Set(event *models.AlertEvent) {
 	}
 }
 
+func (s *SafeEventMap) Init() {
+	aes, err := models.AlertEventGetAll()
+	if err != nil {
+		fmt.Println("load all alert_event fail:", err)
+		os.Exit(1)
+	}
+
+	if len(aes) == 0 {
+		return
+	}
+
+	data := make(map[int64]map[string]*models.AlertEvent)
+	for i := 0; i < len(aes); i++ {
+		event := aes[i]
+		_, has := data[event.RuleId]
+		if !has {
+			m := make(map[string]*models.AlertEvent)
+			m[event.HashId] = event
+			data[event.RuleId] = m
+		} else {
+			data[event.RuleId][event.HashId] = event
+		}
+	}
+
+	s.Lock()
+	s.M = data
+	s.Unlock()
+}
+
 func (s *SafeEventMap) Del(ruleId int64, hashId string) {
 	s.Lock()
 	defer s.Unlock()

models/alert_event.go

@@ -269,6 +269,20 @@ func AlertEventGet(where string, args ...interface{}) (*AlertEvent, error) {
 	return &obj, nil
 }
 
+func AlertEventGetAll() ([]*AlertEvent, error) {
+	var objs []*AlertEvent
+	err := DB.Find(&objs)
+	if err != nil {
+		return objs, err
+	}
+
+	if len(objs) == 0 {
+		return []*AlertEvent{}, nil
+	}
+
+	return objs, nil
+}
+
 // func AlertEventUpdateEventNote(id int64, hashId string, note string, uid int64) error {
 // 	session := DB.NewSession()
 // 	defer session.Close()