update 历史告警控制

consumers/AXY_SK/dataModel.go

@@ -4,6 +4,7 @@ import (
 	"github.com/lib/pq"
 	_ "github.com/lib/pq"
 	"goInOut/models"
+	"time"
 )
 
 type AlarmTrigger struct {
@@ -26,3 +27,10 @@ type StationAlarmGroup struct {
 	Alarm3007 *models.EsAlarm
 	Alarm3008 *models.EsAlarm
 }
+
+type HistoryAlarm struct {
+	SourceId   string
+	Condition  string
+	Time       time.Time
+	AlarmLevel int
+}

consumers/consumerAxySkAlarm.go

@@ -31,6 +31,7 @@ type consumerAxySkAlarm struct {
 	//数据库配置信息
 	stationAlarmTrigger    []AXY_SK.StationAlarmTrigger
 	configAlarmTrigger     []AXY_SK.AlarmTrigger
+	historyStationAlarmMap map[string]AXY_SK.HistoryAlarm
 }
 
 func (the *consumerAxySkAlarm) LoadConfigJson(cfgStr string) {
@@ -59,6 +60,7 @@ func (the *consumerAxySkAlarm) Initial(cfg string) error {
 	if err != nil {
 		return err
 	}
+	the.loadHistoryRecord()
 
 	err = the.monitorInitial()
 	return err
@@ -95,6 +97,10 @@ func (the *consumerAxySkAlarm) infoComponentInitial() error {
 	return nil
 }
 
+func (the *consumerAxySkAlarm) loadHistoryRecord() {
+	the.historyStationAlarmMap = map[string]AXY_SK.HistoryAlarm{}
+}
+
 func (the *consumerAxySkAlarm) monitorInitial() error {
 	the.monitor = &monitors.CommonMonitor{
 		MonitorHelper: &monitors.MonitorHelper{},
@@ -151,7 +157,8 @@ func (the *consumerAxySkAlarm) updateTriggerStationConfig() {
 }
 
 func (the *consumerAxySkAlarm) judgeSK() string {
-	previousTriggerStationAlarmMap := map[string]string{} //同factorId_stationId 触发了高等级 不触发低等级(查询已排序)
+	onceTriggerStationAlarmMap := map[string]string{} //同factorId_stationId 触发了高等级 不触发低等级(查询已排序)
+
 	for i, trigger := range the.configAlarmTrigger {
 		log.Printf("开始处理--> 第[%d] id=%d trigger配置", i, trigger.Id)
 		//配置的结构物的监测因素 去查询
@@ -181,7 +188,7 @@ func (the *consumerAxySkAlarm) judgeSK() string {
 		//判断是否满足告警
 		for sid, stationAlarmInfo := range stationAlarmMap {
 			log.Printf("判断测点[%s] 是否满足双控告警", sid)
-			if v, ok := previousTriggerStationAlarmMap[sid]; ok {
+			if v, ok := onceTriggerStationAlarmMap[sid]; ok {
 				log.Printf("测点[%s]本次已经触发过[%s],不再重复触发", sid, v)
 				continue
 			}
@@ -195,9 +202,33 @@ func (the *consumerAxySkAlarm) judgeSK() string {
 				alarmInfoTemplate = stationAlarmInfo.Alarm3008
 			}
 			if isAlarm && alarmInfoTemplate != nil {
-				previousTriggerStationAlarmMap[sid] = fmt.Sprintf("st:%d,f:%d,level:%d",
+				conditionStr := fmt.Sprintf("st:%d,f:%d,level:%d",
 					trigger.StructId, trigger.FactorId, level)
-				payload := the.skAlarmInfo(alarmInfoTemplate, level, detail)
+				onceTriggerStationAlarmMap[sid] = conditionStr
+				//纪录历史告警
+
+				payload, now := the.skAlarmInfo(alarmInfoTemplate, level, detail)
+				//判断历史有没有
+				if v, ok := the.historyStationAlarmMap[sid]; ok {
+					if v.AlarmLevel > level { //低等级告警过滤
+						log.Printf("测点[%s]本次触发 level=[%d] > 历史有效等级%d,不再重复触发", sid, v.AlarmLevel, level)
+						continue
+					}
+
+					if !v.Time.After(now) {
+						log.Printf("测点[%s]本次触发时刻[%s] 对比历史有效时刻[%s] 非新", sid,
+							v.Time.Format("2006-01-02 15:04:05"), now.Format("2006-01-02 15:04:05"))
+						continue
+					}
+
+				}
+				the.historyStationAlarmMap[sid] = AXY_SK.HistoryAlarm{
+					SourceId:   sid,
+					Condition:  conditionStr,
+					AlarmLevel: level,
+					Time:       now,
+				}
+
 				the.InKafka.Publish(the.Info.IoConfig.In.Kafka.AlarmTopic, payload)
 			} else {
 				payload := the.skAlarmElimination(alarmInfoTemplate, level, detail)
@@ -208,7 +239,8 @@ func (the *consumerAxySkAlarm) judgeSK() string {
 	return ""
 }
 
-func (the *consumerAxySkAlarm) skAlarmInfo(alarmInfoTemplate *models.EsAlarm, level int, detail string) []byte {
+func (the *consumerAxySkAlarm) skAlarmInfo(alarmInfoTemplate *models.EsAlarm, level int, detail string) ([]byte, time.Time) {
+	now := time.Now()
 	alarmMsg := models.KafkaAlarm{
 		MessageMode:   "AlarmGeneration",
 		StructureId:   alarmInfoTemplate.StructureId,
@@ -218,14 +250,14 @@ func (the *consumerAxySkAlarm) skAlarmInfo(alarmInfoTemplate *models.EsAlarm, le
 		AlarmTypeCode: "3077",
 		AlarmCode:     fmt.Sprintf("3077000%d", level),
 		Content:       detail,
-		Time:          time.Now().Format("2006-01-02T15:04:05+0800"),
+		Time:          now.Format("2006-01-02T15:04:05+0800"),
 		SourceTypeId:  2, //   0:DTU, 1:传感器, 2:测点
 		Sponsor:       "goInOut_axySkAlarm",
 		Extras:        nil,
 		SubDevices:    nil,
 	}
 	payload, _ := json.Marshal(alarmMsg)
-	return payload
+	return payload, now
 }
 func (the *consumerAxySkAlarm) skAlarmElimination(alarmInfoTemplate *models.EsAlarm, level int, detail string) []byte {
 	alarmMsg := models.KafkaAlarm{