🎉 Support recover alarm event get current latest value (#152)

Author: lbzss

alert/eval/eval.go

@@ -23,7 +23,7 @@ type (
 		Submit(rule models.AlertRule)
 		Stop(ruleId string)
 		Eval(ctx context.Context, rule models.AlertRule)
-		Recover(ruleId, faultCenterKey string, faultCenterInfoKey string, curKeys []string)
+		Recover(ruleId, faultCenterKey string, faultCenterInfoKey string, curKeys []string, curMetrics map[string]interface{})
 		RestartAllEvals()
 	}
 
@@ -82,6 +82,7 @@ func (t *AlertRule) Eval(ctx context.Context, rule models.AlertRule) {
 			}
 
 			var curFingerprints []string
+			var fingerPrintMetrics map[string]interface{}
 			for _, dsId := range rule.DatasourceIdList {
 				instance, err := t.ctx.DB.Datasource().GetInstance(dsId)
 				if err != nil {
@@ -95,9 +96,10 @@ func (t *AlertRule) Eval(ctx context.Context, rule models.AlertRule) {
 				}
 
 				var fingerprints []string
+
 				switch rule.DatasourceType {
 				case "Prometheus", "VictoriaMetrics":
-					fingerprints = metrics(t.ctx, dsId, instance.Type, rule)
+					fingerprints, fingerPrintMetrics = metrics(t.ctx, dsId, instance.Type, rule)
 				case "AliCloudSLS", "Loki", "ElasticSearch":
 					fingerprints = logs(t.ctx, dsId, instance.Type, rule)
 				case "Jaeger":
@@ -113,7 +115,7 @@ func (t *AlertRule) Eval(ctx context.Context, rule models.AlertRule) {
 				curFingerprints = append(curFingerprints, fingerprints...)
 			}
 			logc.Infof(t.ctx.Ctx, fmt.Sprintf("规则评估 -> %v", tools.JsonMarshal(rule)))
-			t.Recover(rule.RuleId, models.BuildCacheEventKey(rule.TenantId, rule.FaultCenterId), models.BuildCacheInfoKey(rule.TenantId, rule.FaultCenterId), curFingerprints)
+			t.Recover(rule.RuleId, models.BuildCacheEventKey(rule.TenantId, rule.FaultCenterId), models.BuildCacheInfoKey(rule.TenantId, rule.FaultCenterId), curFingerprints, fingerPrintMetrics)
 			t.GC(rule, curFingerprints)
 
 		case <-ctx.Done():
@@ -134,7 +136,7 @@ func (t *AlertRule) getEvalTimeDuration(evalTimeType string, evalInterval int64)
 	}
 }
 
-func (t *AlertRule) Recover(RuleId, faultCenterKey string, faultCenterInfoKey string, curFingerprints []string) {
+func (t *AlertRule) Recover(RuleId, faultCenterKey string, faultCenterInfoKey string, curFingerprints []string, fingerPrintMetrics map[string]interface{}) {
 	// 获取所有的故障中心的告警事件
 	events, err := t.ctx.Redis.Event().GetAllEventsForFaultCenter(faultCenterKey)
 	if err != nil {
@@ -191,7 +193,12 @@ func (t *AlertRule) Recover(RuleId, faultCenterKey string, faultCenterInfoKey st
 		event.IsRecovered = true
 		event.RecoverTime = curTime
 		event.LastSendTime = 0
-		event.Metric["value"] = ""
+		currentMetrics := fingerPrintMetrics[fingerprint]
+		if metric, ok := currentMetrics.(map[string]interface{}); ok {
+			event.Metric["value"] = metric["value"]
+		} else {
+			event.Metric["value"] = 0
+		}
 
 		t.ctx.Redis.Event().PushEventToFaultCenter(&event)
 		// 触发恢复删除带恢复中的 key

alert/eval/query.go

@@ -15,51 +15,54 @@ import (
 )
 
 // Metrics 包含 Prometheus、VictoriaMetrics 数据源
-func metrics(ctx *ctx.Context, datasourceId, datasourceType string, rule models.AlertRule) []string {
+func metrics(ctx *ctx.Context, datasourceId, datasourceType string, rule models.AlertRule) ([]string, map[string]interface{}) {
 	pools := ctx.Redis.ProviderPools()
 	var (
 		resQuery       []provider.Metrics
 		externalLabels map[string]interface{}
 	)
-
+	fingerPrintMetrics := make(map[string]interface{})
 	switch datasourceType {
 	case provider.PrometheusDsProvider:
 		cli, err := pools.GetClient(datasourceId)
 		if err != nil {
 			logc.Errorf(ctx.Ctx, err.Error())
-			return nil
+			return nil, nil
 		}
 
 		resQuery, err = cli.(provider.PrometheusProvider).Query(rule.PrometheusConfig.PromQL)
 		if err != nil {
 			logc.Error(ctx.Ctx, err.Error())
-			return nil
+			return nil, nil
 		}
 
 		externalLabels = cli.(provider.PrometheusProvider).GetExternalLabels()
 	case provider.VictoriaMetricsDsProvider:
 		cli, err := pools.GetClient(datasourceId)
 		if err != nil {
 			logc.Errorf(ctx.Ctx, err.Error())
-			return nil
+			return nil, nil
 		}
 
 		resQuery, err = cli.(provider.VictoriaMetricsProvider).Query(rule.PrometheusConfig.PromQL)
 		if err != nil {
 			logc.Error(ctx.Ctx, err.Error())
-			return nil
+			return nil, nil
 		}
 
 		externalLabels = cli.(provider.VictoriaMetricsProvider).GetExternalLabels()
 	default:
 		logc.Errorf(ctx.Ctx, fmt.Sprintf("Unsupported metrics type, type: %s", datasourceType))
-		return nil
+		return nil, nil
 	}
 
 	if resQuery == nil {
-		return nil
+		return nil, nil
 	}
 
+	// 获取已缓存事件指纹
+	fingerPrintMap := process.GetFingerPrint(ctx, rule.TenantId, rule.FaultCenterId, rule.RuleId)
+
 	var curFingerprints []string
 	for _, v := range resQuery {
 		for _, ruleExpr := range rule.PrometheusConfig.Rules {
@@ -69,36 +72,39 @@ func metrics(ctx *ctx.Context, datasourceId, datasourceType string, rule models.
 				continue
 			}
 
-			event := func() *models.AlertCurEvent {
-				event := process.BuildEvent(rule)
-				event.DatasourceId = datasourceId
-				event.Fingerprint = v.GetFingerprint()
-				event.Metric = v.GetMetric()
-				event.Metric["severity"] = ruleExpr.Severity
-				for ek, ev := range externalLabels {
-					event.Metric[ek] = ev
-				}
-				event.Severity = ruleExpr.Severity
-				event.Annotations = tools.ParserVariables(rule.PrometheusConfig.Annotations, event.Metric)
-				event.SearchQL = rule.PrometheusConfig.PromQL
-				curFingerprints = append(curFingerprints, event.Fingerprint)
-
-				return &event
-			}
-
 			option := models.EvalCondition{
 				Operator:      operator,
 				QueryValue:    v.Value,
 				ExpectedValue: value,
 			}
 
+			event := process.BuildEvent(rule)
+			event.DatasourceId = datasourceId
+			event.Fingerprint = v.GetFingerprint()
+			event.Metric = v.GetMetric()
+			event.Metric["severity"] = ruleExpr.Severity
+			for ek, ev := range externalLabels {
+				event.Metric[ek] = ev
+			}
+			event.Severity = ruleExpr.Severity
+			event.Annotations = tools.ParserVariables(rule.PrometheusConfig.Annotations, event.Metric)
+			event.SearchQL = rule.PrometheusConfig.PromQL
+
 			if process.EvalCondition(option) {
-				process.PushEventToFaultCenter(ctx, event())
+				// 如果告警条件满足需要将告警事件指纹加入，不满足时应当直接跳过
+				curFingerprints = append(curFingerprints, event.Fingerprint)
+
+				process.PushEventToFaultCenter(ctx, &event)
+			} else {
+				// 仅更新已经触发事件的指纹对应指标
+				if _, exist := fingerPrintMap[event.Fingerprint]; exist {
+					fingerPrintMetrics[event.Fingerprint] = event.Metric
+				}
 			}
 		}
 	}
 
-	return curFingerprints
+	return curFingerprints, fingerPrintMetrics
 }
 
 // Logs 包含 AliSLS、Loki、ElasticSearch 数据源

alert/process/process.go

@@ -169,3 +169,13 @@ func RecordAlertHisEvent(ctx *ctx.Context, alert models.AlertCurEvent) error {
 
 	return nil
 }
+
+// GetFingerPrint 获取指纹信息
+func GetFingerPrint(ctx *ctx.Context, tenantId string, faultCenterId string, ruleId string) map[string]struct{} {
+	fingerPrints := ctx.Redis.Event().GetFingerprintsByRuleId(tenantId, faultCenterId, ruleId)
+	fingerPrintMap := make(map[string]struct{})
+	for _, fingerPrint := range fingerPrints {
+		fingerPrintMap[fingerPrint] = struct{}{}
+	}
+	return fingerPrintMap
+}