supplement noti records for agg events

alert/sender/callback.go

@@ -129,34 +129,39 @@ func (c *DefaultCallBacker) CallBack(ctx CallBackContext) {
 		return
 	}
 
-	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, event, "callback", ctx.Stats, event)
+	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, event, "callback", ctx.Stats, ctx.Events)
 }
 
 func doSendAndRecord(ctx *ctx.Context, url, token string, body interface{}, channel string,
-	stats *astats.Stats, event *models.AlertCurEvent) {
+	stats *astats.Stats, events []*models.AlertCurEvent) {
 	res, err := doSend(url, body, channel, stats)
-	NotifyRecord(ctx, event, channel, token, res, err)
+	NotifyRecord(ctx, events, channel, token, res, err)
 }
 
-func NotifyRecord(ctx *ctx.Context, evt *models.AlertCurEvent, channel, target, res string, err error) {
-	noti := models.NewNotificationRecord(evt, channel, target)
-	if err != nil {
-		noti.SetStatus(models.NotiStatusFailure)
-		noti.SetDetails(err.Error())
-	} else if res != "" {
-		noti.SetDetails(string(res))
+func NotifyRecord(ctx *ctx.Context, evts []*models.AlertCurEvent, channel, target, res string, err error) {
+	// 一个通知可能对应多个 event，都需要记录
+	notis := make([]*models.NotificaitonRecord, 0, len(evts))
+	for _, evt := range evts {
+		noti := models.NewNotificationRecord(evt, channel, target)
+		if err != nil {
+			noti.SetStatus(models.NotiStatusFailure)
+			noti.SetDetails(err.Error())
+		} else if res != "" {
+			noti.SetDetails(string(res))
+		}
+		notis = append(notis, noti)
 	}
 
 	if !ctx.IsCenter {
-		_, err := poster.PostByUrlsWithResp[int64](ctx, "/v1/n9e/notify-record", noti)
+		_, err := poster.PostByUrlsWithResp[int64](ctx, "/v1/n9e/notify-record", notis)
 		if err != nil {
-			logger.Errorf("add noti:%v failed, err: %v", noti, err)
+			logger.Errorf("add notis:%v failed, err: %v", notis, err)
 		}
 		return
 	}
 
-	if err := noti.Add(ctx); err != nil {
-		logger.Errorf("add noti:%v failed, err: %v", noti, err)
+	if err := models.DB(ctx).CreateInBatches(notis, 100).Error; err != nil {
+		logger.Errorf("add notis:%v failed, err: %v", notis, err)
 	}
 }
 

alert/sender/dingtalk.go

@@ -67,7 +67,7 @@ func (ds *DingtalkSender) Send(ctx MessageContext) {
 			}
 		}
 
-		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Dingtalk, ctx.Stats, ctx.Events[0])
+		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Dingtalk, ctx.Stats, ctx.Events)
 	}
 }
 
@@ -97,8 +97,7 @@ func (ds *DingtalkSender) CallBack(ctx CallBackContext) {
 		body.Markdown.Text = message
 	}
 
-	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body,
-		"callback", ctx.Stats, ctx.Events[0])
+	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback", ctx.Stats, ctx.Events)
 }
 
 // extract urls and ats from Users

alert/sender/email.go

@@ -25,8 +25,8 @@ type EmailSender struct {
 }
 
 type EmailContext struct {
-	event *models.AlertCurEvent
-	mail  *gomail.Message
+	events []*models.AlertCurEvent
+	mail   *gomail.Message
 }
 
 func (es *EmailSender) Send(ctx MessageContext) {
@@ -42,7 +42,7 @@ func (es *EmailSender) Send(ctx MessageContext) {
 		subject = ctx.Events[0].RuleName
 	}
 	content := BuildTplMessage(models.Email, es.contentTpl, ctx.Events)
-	es.WriteEmail(subject, content, tos, ctx.Events[0])
+	es.WriteEmail(subject, content, tos, ctx.Events)
 
 	ctx.Stats.AlertNotifyTotal.WithLabelValues(models.Email).Add(float64(len(tos)))
 }
@@ -79,16 +79,15 @@ func SendEmail(subject, content string, tos []string, stmp aconf.SMTPConfig) err
 	return nil
 }
 
-func (es *EmailSender) WriteEmail(subject, content string, tos []string,
-	event *models.AlertCurEvent) {
+func (es *EmailSender) WriteEmail(subject, content string, tos []string, events []*models.AlertCurEvent) {
 	m := gomail.NewMessage()
 
 	m.SetHeader("From", es.smtp.From)
 	m.SetHeader("To", tos...)
 	m.SetHeader("Subject", subject)
 	m.SetBody("text/html", content)
 
-	mailch <- &EmailContext{event, m}
+	mailch <- &EmailContext{events, m}
 }
 
 func dialSmtp(d *gomail.Dialer) gomail.SendCloser {
@@ -206,7 +205,7 @@ func startEmailSender(ctx *ctx.Context, smtp aconf.SMTPConfig) {
 				if err == nil {
 					msg = "ok"
 				}
-				NotifyRecord(ctx, m.event, models.Email, to, msg, err)
+				NotifyRecord(ctx, m.events, models.Email, to, msg, err)
 			}
 
 			size++

alert/sender/feishu.go

@@ -54,8 +54,7 @@ func (fs *FeishuSender) CallBack(ctx CallBackContext) {
 		},
 	}
 
-	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback",
-		ctx.Stats, ctx.Events[0])
+	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback", ctx.Stats, ctx.Events)
 }
 
 func (fs *FeishuSender) Send(ctx MessageContext) {
@@ -77,7 +76,7 @@ func (fs *FeishuSender) Send(ctx MessageContext) {
 				IsAtAll:   false,
 			}
 		}
-		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Feishu, ctx.Stats, ctx.Events[0])
+		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Feishu, ctx.Stats, ctx.Events)
 	}
 }
 

alert/sender/feishucard.go

@@ -135,8 +135,7 @@ func (fs *FeishuCardSender) CallBack(ctx CallBackContext) {
 	}
 	parsedURL.RawQuery = ""
 
-	doSendAndRecord(ctx.Ctx, parsedURL.String(), parsedURL.String(), body, "callback",
-		ctx.Stats, ctx.Events[0])
+	doSendAndRecord(ctx.Ctx, parsedURL.String(), parsedURL.String(), body, "callback", ctx.Stats, ctx.Events)
 }
 
 func (fs *FeishuCardSender) Send(ctx MessageContext) {
@@ -160,8 +159,7 @@ func (fs *FeishuCardSender) Send(ctx MessageContext) {
 	body.Card.Elements[0].Text.Content = message
 	body.Card.Elements[2].Elements[0].Content = SendTitle
 	for i, url := range urls {
-		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.FeishuCard,
-			ctx.Stats, ctx.Events[0])
+		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.FeishuCard, ctx.Stats, ctx.Events)
 	}
 }
 

alert/sender/lark.go

@@ -27,8 +27,7 @@ func (lk *LarkSender) CallBack(ctx CallBackContext) {
 		},
 	}
 
-	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback",
-		ctx.Stats, ctx.Events[0])
+	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback", ctx.Stats, ctx.Events)
 }
 
 func (lk *LarkSender) Send(ctx MessageContext) {
@@ -44,7 +43,7 @@ func (lk *LarkSender) Send(ctx MessageContext) {
 				Text: message,
 			},
 		}
-		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Lark, ctx.Stats, ctx.Events[0])
+		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Lark, ctx.Stats, ctx.Events)
 	}
 }
 

alert/sender/larkcard.go

@@ -56,8 +56,7 @@ func (fs *LarkCardSender) CallBack(ctx CallBackContext) {
 	}
 	parsedURL.RawQuery = ""
 
-	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback",
-		ctx.Stats, ctx.Events[0])
+	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback", ctx.Stats, ctx.Events)
 }
 
 func (fs *LarkCardSender) Send(ctx MessageContext) {
@@ -81,7 +80,7 @@ func (fs *LarkCardSender) Send(ctx MessageContext) {
 	body.Card.Elements[0].Text.Content = message
 	body.Card.Elements[2].Elements[0].Content = SendTitle
 	for i, url := range urls {
-		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.LarkCard, ctx.Stats, ctx.Events[0])
+		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.LarkCard, ctx.Stats, ctx.Events)
 	}
 }
 

alert/sender/mm.go

@@ -43,7 +43,7 @@ func (ms *MmSender) Send(ctx MessageContext) {
 		Text:   message,
 		Tokens: urls,
 		Stats:  ctx.Stats,
-	}, ctx.Events[0], models.Mm)
+	}, ctx.Events, models.Mm)
 }
 
 func (ms *MmSender) CallBack(ctx CallBackContext) {
@@ -56,7 +56,7 @@ func (ms *MmSender) CallBack(ctx CallBackContext) {
 		Text:   message,
 		Tokens: []string{ctx.CallBackURL},
 		Stats:  ctx.Stats,
-	}, ctx.Events[0], "callback")
+	}, ctx.Events, "callback")
 }
 
 func (ms *MmSender) extract(users []*models.User) []string {
@@ -69,12 +69,12 @@ func (ms *MmSender) extract(users []*models.User) []string {
 	return tokens
 }
 
-func SendMM(ctx *ctx.Context, message MatterMostMessage, event *models.AlertCurEvent, channel string) {
+func SendMM(ctx *ctx.Context, message MatterMostMessage, events []*models.AlertCurEvent, channel string) {
 	for i := 0; i < len(message.Tokens); i++ {
 		u, err := url.Parse(message.Tokens[i])
 		if err != nil {
 			logger.Errorf("mm_sender: failed to parse error=%v", err)
-			NotifyRecord(ctx, event, channel, message.Tokens[i], "", err)
+			NotifyRecord(ctx, events, channel, message.Tokens[i], "", err)
 			continue
 		}
 
@@ -103,7 +103,7 @@ func SendMM(ctx *ctx.Context, message MatterMostMessage, event *models.AlertCurE
 				Username: username,
 				Text:     txt + message.Text,
 			}
-			doSendAndRecord(ctx, ur, message.Tokens[i], body, channel, message.Stats, event)
+			doSendAndRecord(ctx, ur, message.Tokens[i], body, channel, message.Stats, events)
 		}
 	}
 }

alert/sender/plugin.go

@@ -85,7 +85,7 @@ func alertingCallScript(ctx *ctx.Context, stdinBytes []byte, notifyScript models
 	}
 
 	err, isTimeout := sys.WrapTimeout(cmd, time.Duration(config.Timeout)*time.Second)
-	NotifyRecord(ctx, event, channel, cmd.String(), "", buildErr(err, isTimeout))
+	NotifyRecord(ctx, []*models.AlertCurEvent{event}, channel, cmd.String(), "", buildErr(err, isTimeout))
 
 	if isTimeout {
 		if err == nil {

alert/sender/telegram.go

@@ -41,7 +41,7 @@ func (ts *TelegramSender) CallBack(ctx CallBackContext) {
 		Text:   message,
 		Tokens: []string{ctx.CallBackURL},
 		Stats:  ctx.Stats,
-	}, ctx.Events[0], "callback")
+	}, ctx.Events, "callback")
 }
 
 func (ts *TelegramSender) Send(ctx MessageContext) {
@@ -55,7 +55,7 @@ func (ts *TelegramSender) Send(ctx MessageContext) {
 		Text:   message,
 		Tokens: tokens,
 		Stats:  ctx.Stats,
-	}, ctx.Events[0], models.Telegram)
+	}, ctx.Events, models.Telegram)
 }
 
 func (ts *TelegramSender) extract(users []*models.User) []string {
@@ -68,11 +68,11 @@ func (ts *TelegramSender) extract(users []*models.User) []string {
 	return tokens
 }
 
-func SendTelegram(ctx *ctx.Context, message TelegramMessage, event *models.AlertCurEvent, channel string) {
+func SendTelegram(ctx *ctx.Context, message TelegramMessage, events []*models.AlertCurEvent, channel string) {
 	for i := 0; i < len(message.Tokens); i++ {
 		if !strings.Contains(message.Tokens[i], "/") && !strings.HasPrefix(message.Tokens[i], "https://") {
 			logger.Errorf("telegram_sender: result=fail invalid token=%s", message.Tokens[i])
-			NotifyRecord(ctx, event, channel, message.Tokens[i], "", errors.New("invalid token"))
+			NotifyRecord(ctx, events, channel, message.Tokens[i], "", errors.New("invalid token"))
 			continue
 		}
 		var url string
@@ -93,6 +93,6 @@ func SendTelegram(ctx *ctx.Context, message TelegramMessage, event *models.Alert
 			Text:      message.Text,
 		}
 
-		doSendAndRecord(ctx, url, message.Tokens[i], body, channel, message.Stats, event)
+		doSendAndRecord(ctx, url, message.Tokens[i], body, channel, message.Stats, events)
 	}
 }

alert/sender/webhook.go

@@ -96,7 +96,7 @@ func SingleSendWebhooks(ctx *ctx.Context, webhooks []*models.Webhook, event *mod
 		retryCount := 0
 		for retryCount < 3 {
 			needRetry, res, err := sendWebhook(conf, event, stats)
-			NotifyRecord(ctx, event, "webhook", conf.Url, res, err)
+			NotifyRecord(ctx, []*models.AlertCurEvent{event}, "webhook", conf.Url, res, err)
 			if !needRetry {
 				break
 			}
@@ -166,7 +166,7 @@ func StartConsumer(ctx *ctx.Context, queue *WebhookQueue, popSize int, webhook *
 			retryCount := 0
 			for retryCount < webhook.RetryCount {
 				needRetry, res, err := sendWebhook(webhook, events, stats)
-				go RecordEvents(ctx, webhook, events, stats, res, err)
+				go NotifyRecord(ctx, events, "webhook", webhook.Url, res, err)
 				if !needRetry {
 					break
 				}
@@ -176,10 +176,3 @@ func StartConsumer(ctx *ctx.Context, queue *WebhookQueue, popSize int, webhook *
 		}
 	}
 }
-
-func RecordEvents(ctx *ctx.Context, webhook *models.Webhook, events []*models.AlertCurEvent, stats *astats.Stats, res string, err error) {
-	for _, event := range events {
-		time.Sleep(time.Millisecond * 10)
-		NotifyRecord(ctx, event, "webhook", webhook.Url, res, err)
-	}
-}

alert/sender/wecom.go

@@ -37,8 +37,7 @@ func (ws *WecomSender) CallBack(ctx CallBackContext) {
 		},
 	}
 
-	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback",
-		ctx.Stats, ctx.Events[0])
+	doSendAndRecord(ctx.Ctx, ctx.CallBackURL, ctx.CallBackURL, body, "callback", ctx.Stats, ctx.Events)
 }
 
 func (ws *WecomSender) Send(ctx MessageContext) {
@@ -54,7 +53,7 @@ func (ws *WecomSender) Send(ctx MessageContext) {
 				Content: message,
 			},
 		}
-		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Wecom, ctx.Stats, ctx.Events[0])
+		doSendAndRecord(ctx.Ctx, url, tokens[i], body, models.Wecom, ctx.Stats, ctx.Events)
 	}
 }
 

center/router/router_notification_record.go

@@ -35,11 +35,18 @@ type Record struct {
 
 // notificationRecordAdd
 func (rt *Router) notificationRecordAdd(c *gin.Context) {
-	var req models.NotificaitonRecord
+	var req []*models.NotificaitonRecord
 	ginx.BindJSON(c, &req)
-	err := req.Add(rt.Ctx)
+	err := models.DB(rt.Ctx).CreateInBatches(req, 100).Error
+	var ids []int64
+	if err == nil {
+		ids = make([]int64, len(req))
+		for i, noti := range req {
+			ids[i] = noti.Id
+		}
+	}
 
-	ginx.NewRender(c).Data(req.Id, err)
+	ginx.NewRender(c).Data(ids, err)
 }
 
 func (rt *Router) notificationRecordList(c *gin.Context) {