add stats

* feat: support lark notify channel (#2056)

Co-authored-by: flashbo <36443248+lwb0214@users.noreply.github.com>
Co-authored-by: wenbo <1027758873@qq.com>

alert/aconf/conf.go

@@ -32,6 +32,7 @@ type Alerting struct {
 	Timeout           int64
 	TemplatesDir      string
 	NotifyConcurrency int
+	WebhookBatchSend  bool
 }
 
 type CallPlugin struct {

alert/dispatch/consume.go

@@ -14,9 +14,7 @@ import (
 	"github.com/ccfos/nightingale/v6/pkg/poster"
 	promsdk "github.com/ccfos/nightingale/v6/pkg/prom"
 	"github.com/ccfos/nightingale/v6/prom"
-	"github.com/ccfos/nightingale/v6/pushgw/writer"
 
-	"github.com/prometheus/prometheus/prompb"
 	"github.com/toolkits/pkg/concurrent/semaphore"
 	"github.com/toolkits/pkg/logger"
 )
@@ -95,8 +93,6 @@ func (e *Consumer) consumeOne(event *models.AlertCurEvent) {
 		event.RuleNote = fmt.Sprintf("failed to parse rule note: %v", err)
 	}
 
-	e.relabel(event)
-
 	e.persist(event)
 
 	if event.IsRecovered && event.NotifyRecovered == 0 {
@@ -190,51 +186,6 @@ func (e *Consumer) queryRecoveryVal(event *models.AlertCurEvent) {
 	}
 }
 
-func (e *Consumer) relabel(event *models.AlertCurEvent) {
-	rule := e.dispatch.alertRuleCache.Get(event.RuleId)
-	if rule == nil {
-		return
-	}
-
-	// need to keep the original label
-	event.OriginalTags = event.Tags
-	event.OriginalTagsJSON = make([]string, len(event.TagsJSON))
-
-	labels := make([]prompb.Label, len(event.TagsJSON))
-	for i, tag := range event.TagsJSON {
-		label := strings.Split(tag, "=")
-		if len(label) != 2 {
-			logger.Errorf("event%+v relabel: the label length is not 2:%v", event, label)
-			continue
-		}
-		event.OriginalTagsJSON[i] = tag
-		labels[i] = prompb.Label{Name: label[0], Value: label[1]}
-	}
-
-	for i := 0; i < len(rule.EventRelabelConfig); i++ {
-		if rule.EventRelabelConfig[i].Replacement == "" {
-			rule.EventRelabelConfig[i].Replacement = "$1"
-		}
-
-		if rule.EventRelabelConfig[i].Separator == "" {
-			rule.EventRelabelConfig[i].Separator = ";"
-		}
-
-		if rule.EventRelabelConfig[i].Regex == "" {
-			rule.EventRelabelConfig[i].Regex = "(.*)"
-		}
-	}
-
-	// relabel process
-	relabels := writer.Process(labels, rule.EventRelabelConfig...)
-	event.TagsJSON = make([]string, len(relabels))
-	for i, label := range relabels {
-		event.TagsJSON[i] = fmt.Sprintf("%s=%s", label.Name, label.Value)
-		event.TagsMap[label.Name] = label.Value
-	}
-	event.Tags = strings.Join(event.TagsJSON, ",,")
-}
-
 func getKey(event *models.AlertCurEvent) string {
 	return common.RuleKey(event.DatasourceId, event.RuleId)
 }

alert/dispatch/dispatch.go

@@ -100,6 +100,8 @@ func (e *Dispatch) relaodTpls() error {
 		models.Mm:         sender.NewSender(models.Mm, tmpTpls),
 		models.Telegram:   sender.NewSender(models.Telegram, tmpTpls),
 		models.FeishuCard: sender.NewSender(models.FeishuCard, tmpTpls),
+		models.Lark:       sender.NewSender(models.Lark, tmpTpls),
+		models.LarkCard:   sender.NewSender(models.LarkCard, tmpTpls),
 	}
 
 	// domain -> Callback()
@@ -110,7 +112,9 @@ func (e *Dispatch) relaodTpls() error {
 		models.TelegramDomain:   sender.NewCallBacker(models.TelegramDomain, e.targetCache, e.userCache, e.taskTplsCache, tmpTpls),
 		models.FeishuCardDomain: sender.NewCallBacker(models.FeishuCardDomain, e.targetCache, e.userCache, e.taskTplsCache, tmpTpls),
 		models.IbexDomain:       sender.NewCallBacker(models.IbexDomain, e.targetCache, e.userCache, e.taskTplsCache, tmpTpls),
+		models.LarkDomain:       sender.NewCallBacker(models.LarkDomain, e.targetCache, e.userCache, e.taskTplsCache, tmpTpls),
 		models.DefaultDomain:    sender.NewCallBacker(models.DefaultDomain, e.targetCache, e.userCache, e.taskTplsCache, tmpTpls),
+		models.LarkCardDomain:   sender.NewCallBacker(models.LarkCardDomain, e.targetCache, e.userCache, e.taskTplsCache, tmpTpls),
 	}
 
 	e.RwLock.RLock()
@@ -261,7 +265,11 @@ func (e *Dispatch) Send(rule *models.AlertRule, event *models.AlertCurEvent, not
 	e.SendCallbacks(rule, notifyTarget, event)
 
 	// handle global webhooks
-	sender.SendWebhooks(notifyTarget.ToWebhookList(), event, e.Astats)
+	if e.alerting.WebhookBatchSend {
+		sender.BatchSendWebhooks(notifyTarget.ToWebhookList(), event, e.Astats)
+	} else {
+		sender.SingleSendWebhooks(notifyTarget.ToWebhookList(), event, e.Astats)
+	}
 
 	// handle plugin call
 	go sender.MayPluginNotify(e.genNoticeBytes(event), e.notifyConfigCache.GetNotifyScript(), e.Astats)
@@ -276,7 +284,7 @@ func (e *Dispatch) SendCallbacks(rule *models.AlertRule, notifyTarget *NotifyTar
 			continue
 		}
 
-		cbCtx := sender.BuildCallBackContext(e.ctx, urlStr, rule, []*models.AlertCurEvent{event}, uids, e.userCache, e.Astats)
+		cbCtx := sender.BuildCallBackContext(e.ctx, urlStr, rule, []*models.AlertCurEvent{event}, uids, e.userCache, e.alerting.WebhookBatchSend, e.Astats)
 
 		if strings.HasPrefix(urlStr, "${ibex}") {
 			e.CallBacks[models.IbexDomain].CallBack(cbCtx)
@@ -299,6 +307,12 @@ func (e *Dispatch) SendCallbacks(rule *models.AlertRule, notifyTarget *NotifyTar
 			continue
 		}
 
+		// process lark card
+		if parsedURL.Host == models.LarkDomain && parsedURL.Query().Get("card") == "1" {
+			e.CallBacks[models.LarkCardDomain].CallBack(cbCtx)
+			continue
+		}
+
 		callBacker, ok := e.CallBacks[parsedURL.Host]
 		if ok {
 			callBacker.CallBack(cbCtx)

alert/dispatch/notify_target.go

@@ -79,6 +79,22 @@ func (s *NotifyTarget) ToCallbackList() []string {
 func (s *NotifyTarget) ToWebhookList() []*models.Webhook {
 	webhooks := make([]*models.Webhook, 0, len(s.webhooks))
 	for _, wh := range s.webhooks {
+		if wh.Batch == 0 {
+			wh.Batch = 1000
+		}
+
+		if wh.Timeout == 0 {
+			wh.Timeout = 10
+		}
+
+		if wh.RetryCount == 0 {
+			wh.RetryCount = 10
+		}
+
+		if wh.RetryInterval == 0 {
+			wh.RetryInterval = 10
+		}
+
 		webhooks = append(webhooks, wh)
 	}
 	return webhooks

alert/process/process.go

@@ -18,6 +18,9 @@ import (
 	"github.com/ccfos/nightingale/v6/models"
 	"github.com/ccfos/nightingale/v6/pkg/ctx"
 	"github.com/ccfos/nightingale/v6/pkg/tplx"
+	"github.com/ccfos/nightingale/v6/pushgw/writer"
+
+	"github.com/prometheus/prometheus/prompb"
 	"github.com/toolkits/pkg/logger"
 	"github.com/toolkits/pkg/str"
 )
@@ -218,9 +221,57 @@ func (p *Processor) BuildEvent(anomalyPoint common.AnomalyPoint, from string, no
 	} else {
 		event.LastEvalTime = event.TriggerTime
 	}
+
+	// 生成事件之后，立马进程 relabel 处理
+	Relabel(p.rule, event)
 	return event
 }
 
+func Relabel(rule *models.AlertRule, event *models.AlertCurEvent) {
+	if rule == nil {
+		return
+	}
+
+	// need to keep the original label
+	event.OriginalTags = event.Tags
+	event.OriginalTagsJSON = make([]string, len(event.TagsJSON))
+
+	labels := make([]prompb.Label, len(event.TagsJSON))
+	for i, tag := range event.TagsJSON {
+		label := strings.Split(tag, "=")
+		if len(label) != 2 {
+			logger.Errorf("event%+v relabel: the label length is not 2:%v", event, label)
+			continue
+		}
+		event.OriginalTagsJSON[i] = tag
+		labels[i] = prompb.Label{Name: label[0], Value: label[1]}
+	}
+
+	for i := 0; i < len(rule.EventRelabelConfig); i++ {
+		if rule.EventRelabelConfig[i].Replacement == "" {
+			rule.EventRelabelConfig[i].Replacement = "$1"
+		}
+
+		if rule.EventRelabelConfig[i].Separator == "" {
+			rule.EventRelabelConfig[i].Separator = ";"
+		}
+
+		if rule.EventRelabelConfig[i].Regex == "" {
+			rule.EventRelabelConfig[i].Regex = "(.*)"
+		}
+	}
+
+	// relabel process
+	relabels := writer.Process(labels, rule.EventRelabelConfig...)
+	event.TagsJSON = make([]string, len(relabels))
+	event.TagsMap = make(map[string]string, len(relabels))
+	for i, label := range relabels {
+		event.TagsJSON[i] = fmt.Sprintf("%s=%s", label.Name, label.Value)
+		event.TagsMap[label.Name] = label.Value
+	}
+	event.Tags = strings.Join(event.TagsJSON, ",,")
+}
+
 func (p *Processor) HandleRecover(alertingKeys map[string]struct{}, now int64, inhibit bool) {
 	for _, hash := range p.pendings.Keys() {
 		if _, has := alertingKeys[hash]; has {

alert/sender/callback.go

@@ -29,13 +29,14 @@ type (
 		Rule        *models.AlertRule
 		Events      []*models.AlertCurEvent
 		Stats       *astats.Stats
+		BatchSend   bool
 	}
 
 	DefaultCallBacker struct{}
 )
 
 func BuildCallBackContext(ctx *ctx.Context, callBackURL string, rule *models.AlertRule, events []*models.AlertCurEvent,
-	uids []int64, userCache *memsto.UserCacheType, stats *astats.Stats) CallBackContext {
+	uids []int64, userCache *memsto.UserCacheType, batchSend bool, stats *astats.Stats) CallBackContext {
 	users := userCache.GetByUserIds(uids)
 
 	newCallBackUrl, _ := events[0].ParseURL(callBackURL)
@@ -45,6 +46,7 @@ func BuildCallBackContext(ctx *ctx.Context, callBackURL string, rule *models.Ale
 		Rule:        rule,
 		Events:      events,
 		Users:       users,
+		BatchSend:   batchSend,
 		Stats:       stats,
 	}
 }
@@ -96,6 +98,10 @@ func NewCallBacker(
 	//	return &MmSender{tpl: tpls[models.Mm]}
 	case models.TelegramDomain:
 		return &TelegramSender{tpl: tpls[models.Telegram]}
+	case models.LarkDomain:
+		return &LarkSender{tpl: tpls[models.Lark]}
+	case models.LarkCardDomain:
+		return &LarkCardSender{tpl: tpls[models.LarkCard]}
 	}
 
 	return nil
@@ -108,6 +114,21 @@ func (c *DefaultCallBacker) CallBack(ctx CallBackContext) {
 
 	event := ctx.Events[0]
 
+	if ctx.BatchSend {
+		webhookConf := &models.Webhook{
+			Type:          models.RuleCallback,
+			Enable:        true,
+			Url:           ctx.CallBackURL,
+			Timeout:       5,
+			RetryCount:    3,
+			RetryInterval: 10,
+			Batch:         1000,
+		}
+
+		PushCallbackEvent(webhookConf, event, ctx.Stats)
+		return
+	}
+
 	ctx.Stats.AlertNotifyTotal.WithLabelValues("rule_callback").Inc()
 	resp, code, err := poster.PostJSON(ctx.CallBackURL, 5*time.Second, event, 3)
 	if err != nil {
@@ -136,3 +157,27 @@ type TaskCreateReply struct {
 	Err string `json:"err"`
 	Dat int64  `json:"dat"` // task.id
 }
+
+func PushCallbackEvent(webhook *models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
+	CallbackEventQueueLock.RLock()
+	queue := CallbackEventQueue[webhook.Url]
+	CallbackEventQueueLock.RUnlock()
+
+	if queue == nil {
+		queue = &WebhookQueue{
+			list:    NewSafeListLimited(QueueMaxSize),
+			closeCh: make(chan struct{}),
+		}
+
+		CallbackEventQueueLock.Lock()
+		CallbackEventQueue[webhook.Url] = queue
+		CallbackEventQueueLock.Unlock()
+
+		StartConsumer(queue, webhook.Batch, webhook, stats)
+	}
+
+	succ := queue.list.PushFront(event)
+	if !succ {
+		logger.Warningf("Write channel(%s) full, current channel size: %d event:%v", webhook.Url, queue.list.Len(), event)
+	}
+}

alert/sender/lark.go

@@ -0,0 +1,64 @@
+package sender
+
+import (
+	"html/template"
+	"strings"
+
+	"github.com/ccfos/nightingale/v6/models"
+)
+
+var (
+	_ CallBacker = (*LarkSender)(nil)
+)
+
+type LarkSender struct {
+	tpl *template.Template
+}
+
+func (lk *LarkSender) CallBack(ctx CallBackContext) {
+	if len(ctx.Events) == 0 || len(ctx.CallBackURL) == 0 {
+		return
+	}
+
+	body := feishu{
+		Msgtype: "text",
+		Content: feishuContent{
+			Text: BuildTplMessage(models.Lark, lk.tpl, ctx.Events),
+		},
+	}
+
+	doSend(ctx.CallBackURL, body, models.Lark, ctx.Stats)
+	ctx.Stats.AlertNotifyTotal.WithLabelValues("rule_callback").Inc()
+}
+
+func (lk *LarkSender) Send(ctx MessageContext) {
+	if len(ctx.Users) == 0 || len(ctx.Events) == 0 {
+		return
+	}
+	urls := lk.extract(ctx.Users)
+	message := BuildTplMessage(models.Lark, lk.tpl, ctx.Events)
+	for _, url := range urls {
+		body := feishu{
+			Msgtype: "text",
+			Content: feishuContent{
+				Text: message,
+			},
+		}
+		doSend(url, body, models.Lark, ctx.Stats)
+	}
+}
+
+func (lk *LarkSender) extract(users []*models.User) []string {
+	urls := make([]string, 0, len(users))
+
+	for _, user := range users {
+		if token, has := user.ExtractToken(models.Lark); has {
+			url := token
+			if !strings.HasPrefix(token, "https://") && !strings.HasPrefix(token, "http://") {
+				url = "https://open.larksuite.com/open-apis/bot/v2/hook/" + token
+			}
+			urls = append(urls, url)
+		}
+	}
+	return urls
+}

alert/sender/larkcard.go

@@ -0,0 +1,98 @@
+package sender
+
+import (
+	"fmt"
+	"html/template"
+	"net/url"
+	"strings"
+
+	"github.com/ccfos/nightingale/v6/models"
+)
+
+type LarkCardSender struct {
+	tpl *template.Template
+}
+
+func (fs *LarkCardSender) CallBack(ctx CallBackContext) {
+	if len(ctx.Events) == 0 || len(ctx.CallBackURL) == 0 {
+		return
+	}
+
+	ats := ExtractAtsParams(ctx.CallBackURL)
+	message := BuildTplMessage(models.LarkCard, fs.tpl, ctx.Events)
+
+	if len(ats) > 0 {
+		atTags := ""
+		for _, at := range ats {
+			if strings.Contains(at, "@") {
+				atTags += fmt.Sprintf("<at email=\"%s\" ></at>", at)
+			} else {
+				atTags += fmt.Sprintf("<at id=\"%s\" ></at>", at)
+			}
+		}
+		message = atTags + message
+	}
+
+	color := "red"
+	lowerUnicode := strings.ToLower(message)
+	if strings.Count(lowerUnicode, Recovered) > 0 && strings.Count(lowerUnicode, Triggered) > 0 {
+		color = "orange"
+	} else if strings.Count(lowerUnicode, Recovered) > 0 {
+		color = "green"
+	}
+
+	SendTitle := fmt.Sprintf("🔔 %s", ctx.Events[0].RuleName)
+	body.Card.Header.Title.Content = SendTitle
+	body.Card.Header.Template = color
+	body.Card.Elements[0].Text.Content = message
+	body.Card.Elements[2].Elements[0].Content = SendTitle
+
+	// This is to be compatible with the Larkcard interface, if with query string parameters, the request will fail
+	// Remove query parameters from the URL,
+	parsedURL, err := url.Parse(ctx.CallBackURL)
+	if err != nil {
+		return
+	}
+	parsedURL.RawQuery = ""
+
+	doSend(parsedURL.String(), body, models.LarkCard, ctx.Stats)
+}
+
+func (fs *LarkCardSender) Send(ctx MessageContext) {
+	if len(ctx.Users) == 0 || len(ctx.Events) == 0 {
+		return
+	}
+	urls, _ := fs.extract(ctx.Users)
+	message := BuildTplMessage(models.LarkCard, fs.tpl, ctx.Events)
+	color := "red"
+	lowerUnicode := strings.ToLower(message)
+	if strings.Count(lowerUnicode, Recovered) > 0 && strings.Count(lowerUnicode, Triggered) > 0 {
+		color = "orange"
+	} else if strings.Count(lowerUnicode, Recovered) > 0 {
+		color = "green"
+	}
+
+	SendTitle := fmt.Sprintf("🔔 %s", ctx.Events[0].RuleName)
+	body.Card.Header.Title.Content = SendTitle
+	body.Card.Header.Template = color
+	body.Card.Elements[0].Text.Content = message
+	body.Card.Elements[2].Elements[0].Content = SendTitle
+	for _, url := range urls {
+		doSend(url, body, models.LarkCard, ctx.Stats)
+	}
+}
+
+func (fs *LarkCardSender) extract(users []*models.User) ([]string, []string) {
+	urls := make([]string, 0, len(users))
+	ats := make([]string, 0)
+	for i := range users {
+		if token, has := users[i].ExtractToken(models.Lark); has {
+			url := token
+			if !strings.HasPrefix(token, "https://") && !strings.HasPrefix(token, "http://") {
+				url = "https://open.larksuite.com/open-apis/bot/v2/hook/" + strings.TrimSpace(token)
+			}
+			urls = append(urls, url)
+		}
+	}
+	return urls, ats
+}

alert/sender/sender.go

@@ -41,6 +41,10 @@ func NewSender(key string, tpls map[string]*template.Template, smtp ...aconf.SMT
 		return &MmSender{tpl: tpls[models.Mm]}
 	case models.Telegram:
 		return &TelegramSender{tpl: tpls[models.Telegram]}
+	case models.Lark:
+		return &LarkSender{tpl: tpls[models.Lark]}
+	case models.LarkCard:
+		return &LarkCardSender{tpl: tpls[models.LarkCard]}
 	}
 	return nil
 }

alert/sender/webhook.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/ccfos/nightingale/v6/alert/astats"
@@ -14,14 +15,19 @@ import (
 	"github.com/toolkits/pkg/logger"
 )
 
-func sendWebhook(webhook *models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) bool {
+func sendWebhook(webhook *models.Webhook, event interface{}, stats *astats.Stats) bool {
+	channel := "webhook"
+	if webhook.Type == models.RuleCallback {
+		channel = "callback"
+	}
+
 	conf := webhook
 	if conf.Url == "" || !conf.Enable {
 		return false
 	}
 	bs, err := json.Marshal(event)
 	if err != nil {
-		logger.Errorf("alertingWebhook failed to marshal event:%+v err:%v", event, err)
+		logger.Errorf("%s alertingWebhook failed to marshal event:%+v err:%v", channel, event, err)
 		return false
 	}
 
@@ -29,7 +35,7 @@ func sendWebhook(webhook *models.Webhook, event *models.AlertCurEvent, stats *as
 
 	req, err := http.NewRequest("POST", conf.Url, bf)
 	if err != nil {
-		logger.Warningf("alertingWebhook failed to new reques event:%+v err:%v", event, err)
+		logger.Warningf("%s alertingWebhook failed to new reques event:%s err:%v", channel, string(bs), err)
 		return true
 	}
 
@@ -58,12 +64,12 @@ func sendWebhook(webhook *models.Webhook, event *models.AlertCurEvent, stats *as
 		},
 	}
 
-	stats.AlertNotifyTotal.WithLabelValues("webhook").Inc()
+	stats.AlertNotifyTotal.WithLabelValues(channel).Inc()
 	var resp *http.Response
 	resp, err = client.Do(req)
 	if err != nil {
-		stats.AlertNotifyErrorTotal.WithLabelValues("webhook").Inc()
-		logger.Errorf("event_webhook_fail, ruleId: [%d], eventId: [%d], event:%+v, url: [%s], error: [%s]", event.RuleId, event.Id, event, conf.Url, err)
+		stats.AlertNotifyErrorTotal.WithLabelValues(channel).Inc()
+		logger.Errorf("event_%s_fail, event:%s, url: [%s], error: [%s]", channel, string(bs), conf.Url, err)
 		return true
 	}
 
@@ -74,15 +80,15 @@ func sendWebhook(webhook *models.Webhook, event *models.AlertCurEvent, stats *as
 	}
 
 	if resp.StatusCode == 429 {
-		logger.Errorf("event_webhook_fail, url: %s, response code: %d, body: %s event:%+v", conf.Url, resp.StatusCode, string(body), event)
+		logger.Errorf("event_%s_fail, url: %s, response code: %d, body: %s event:%s", channel, conf.Url, resp.StatusCode, string(body), string(bs))
 		return true
 	}
 
-	logger.Debugf("event_webhook_succ, url: %s, response code: %d, body: %s event:%+v", conf.Url, resp.StatusCode, string(body), event)
+	logger.Debugf("event_%s_succ, url: %s, response code: %d, body: %s event:%s", channel, conf.Url, resp.StatusCode, string(body), string(bs))
 	return false
 }
 
-func SendWebhooks(webhooks []*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
+func SingleSendWebhooks(webhooks []*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
 	for _, conf := range webhooks {
 		retryCount := 0
 		for retryCount < 3 {
@@ -95,3 +101,73 @@ func SendWebhooks(webhooks []*models.Webhook, event *models.AlertCurEvent, stats
 		}
 	}
 }
+
+func BatchSendWebhooks(webhooks []*models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
+	for _, conf := range webhooks {
+		logger.Infof("push event:%+v to queue:%v", event, conf)
+		PushEvent(conf, event, stats)
+	}
+}
+
+var EventQueue = make(map[string]*WebhookQueue)
+var CallbackEventQueue = make(map[string]*WebhookQueue)
+var CallbackEventQueueLock sync.RWMutex
+var EventQueueLock sync.RWMutex
+
+const QueueMaxSize = 100000
+
+type WebhookQueue struct {
+	list    *SafeListLimited
+	closeCh chan struct{}
+}
+
+func PushEvent(webhook *models.Webhook, event *models.AlertCurEvent, stats *astats.Stats) {
+	EventQueueLock.RLock()
+	queue := EventQueue[webhook.Url]
+	EventQueueLock.RUnlock()
+
+	if queue == nil {
+		queue = &WebhookQueue{
+			list:    NewSafeListLimited(QueueMaxSize),
+			closeCh: make(chan struct{}),
+		}
+
+		EventQueueLock.Lock()
+		EventQueue[webhook.Url] = queue
+		EventQueueLock.Unlock()
+
+		StartConsumer(queue, webhook.Batch, webhook, stats)
+	}
+
+	succ := queue.list.PushFront(event)
+	if !succ {
+		stats.AlertNotifyErrorTotal.WithLabelValues("push_event_queue").Inc()
+		logger.Warningf("Write channel(%s) full, current channel size: %d event:%v", webhook.Url, queue.list.Len(), event)
+	}
+}
+
+func StartConsumer(queue *WebhookQueue, popSize int, webhook *models.Webhook, stats *astats.Stats) {
+	for {
+		select {
+		case <-queue.closeCh:
+			logger.Infof("event queue:%v closed", queue)
+			return
+		default:
+			events := queue.list.PopBack(popSize)
+			if len(events) == 0 {
+				time.Sleep(time.Millisecond * 400)
+				continue
+			}
+
+			retryCount := 0
+			for retryCount < webhook.RetryCount {
+				needRetry := sendWebhook(webhook, events, stats)
+				if !needRetry {
+					break
+				}
+				retryCount++
+				time.Sleep(time.Second * time.Duration(webhook.RetryInterval) * time.Duration(retryCount))
+			}
+		}
+	}
+}

alert/sender/webhook_queue.go

@@ -0,0 +1,111 @@
+package sender
+
+import (
+	"container/list"
+	"sync"
+
+	"github.com/ccfos/nightingale/v6/models"
+)
+
+type SafeList struct {
+	sync.RWMutex
+	L *list.List
+}
+
+func NewSafeList() *SafeList {
+	return &SafeList{L: list.New()}
+}
+
+func (sl *SafeList) PushFront(v interface{}) *list.Element {
+	sl.Lock()
+	e := sl.L.PushFront(v)
+	sl.Unlock()
+	return e
+}
+
+func (sl *SafeList) PushFrontBatch(vs []interface{}) {
+	sl.Lock()
+	for _, item := range vs {
+		sl.L.PushFront(item)
+	}
+	sl.Unlock()
+}
+
+func (sl *SafeList) PopBack(max int) []*models.AlertCurEvent {
+	sl.Lock()
+
+	count := sl.L.Len()
+	if count == 0 {
+		sl.Unlock()
+		return []*models.AlertCurEvent{}
+	}
+
+	if count > max {
+		count = max
+	}
+
+	items := make([]*models.AlertCurEvent, 0, count)
+	for i := 0; i < count; i++ {
+		item := sl.L.Remove(sl.L.Back())
+		sample, ok := item.(*models.AlertCurEvent)
+		if ok {
+			items = append(items, sample)
+		}
+	}
+
+	sl.Unlock()
+	return items
+}
+
+func (sl *SafeList) RemoveAll() {
+	sl.Lock()
+	sl.L.Init()
+	sl.Unlock()
+}
+
+func (sl *SafeList) Len() int {
+	sl.RLock()
+	size := sl.L.Len()
+	sl.RUnlock()
+	return size
+}
+
+// SafeList with Limited Size
+type SafeListLimited struct {
+	maxSize int
+	SL      *SafeList
+}
+
+func NewSafeListLimited(maxSize int) *SafeListLimited {
+	return &SafeListLimited{SL: NewSafeList(), maxSize: maxSize}
+}
+
+func (sll *SafeListLimited) PopBack(max int) []*models.AlertCurEvent {
+	return sll.SL.PopBack(max)
+}
+
+func (sll *SafeListLimited) PushFront(v interface{}) bool {
+	if sll.SL.Len() >= sll.maxSize {
+		return false
+	}
+
+	sll.SL.PushFront(v)
+	return true
+}
+
+func (sll *SafeListLimited) PushFrontBatch(vs []interface{}) bool {
+	if sll.SL.Len() >= sll.maxSize {
+		return false
+	}
+
+	sll.SL.PushFrontBatch(vs)
+	return true
+}
+
+func (sll *SafeListLimited) RemoveAll() {
+	sll.SL.RemoveAll()
+}
+
+func (sll *SafeListLimited) Len() int {
+	return sll.SL.Len()
+}

center/center.go

@@ -107,7 +107,7 @@ func Initialize(configDir string, cryptoKey string) (func(), error) {
 	go version.GetGithubVersion()
 
 	alertrtRouter := alertrt.New(config.HTTP, config.Alert, alertMuteCache, targetCache, busiGroupCache, alertStats, ctx, externalProcessors)
-	centerRouter := centerrt.New(config.HTTP, config.Center, config.Alert, cconf.Operations, dsCache, notifyConfigCache, promClients, tdengineClients,
+	centerRouter := centerrt.New(config.HTTP, config.Center, config.Alert, config.Ibex, cconf.Operations, dsCache, notifyConfigCache, promClients, tdengineClients,
 		redis, sso, ctx, metas, idents, targetCache, userCache, userGroupCache)
 	pushgwRouter := pushgwrt.New(config.HTTP, config.Pushgw, config.Alert, targetCache, busiGroupCache, idents, metas, writers, ctx)
 

center/router/router.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ccfos/nightingale/v6/center/cstats"
 	"github.com/ccfos/nightingale/v6/center/metas"
 	"github.com/ccfos/nightingale/v6/center/sso"
+	"github.com/ccfos/nightingale/v6/conf"
 	_ "github.com/ccfos/nightingale/v6/front/statik"
 	"github.com/ccfos/nightingale/v6/memsto"
 	"github.com/ccfos/nightingale/v6/pkg/aop"
@@ -34,6 +35,7 @@ import (
 type Router struct {
 	HTTP              httpx.Config
 	Center            cconf.Center
+	Ibex              conf.Ibex
 	Alert             aconf.Alert
 	Operations        cconf.Operation
 	DatasourceCache   *memsto.DatasourceCacheType
@@ -48,13 +50,15 @@ type Router struct {
 	UserCache         *memsto.UserCacheType
 	UserGroupCache    *memsto.UserGroupCacheType
 	Ctx               *ctx.Context
+	HeartbeatHook     HeartbeatHookFunc
 }
 
-func New(httpConfig httpx.Config, center cconf.Center, alert aconf.Alert, operations cconf.Operation, ds *memsto.DatasourceCacheType, ncc *memsto.NotifyConfigCacheType, pc *prom.PromClientMap, tdendgineClients *tdengine.TdengineClientMap, redis storage.Redis, sso *sso.SsoClient, ctx *ctx.Context, metaSet *metas.Set, idents *idents.Set, tc *memsto.TargetCacheType, uc *memsto.UserCacheType, ugc *memsto.UserGroupCacheType) *Router {
+func New(httpConfig httpx.Config, center cconf.Center, alert aconf.Alert, ibex conf.Ibex, operations cconf.Operation, ds *memsto.DatasourceCacheType, ncc *memsto.NotifyConfigCacheType, pc *prom.PromClientMap, tdendgineClients *tdengine.TdengineClientMap, redis storage.Redis, sso *sso.SsoClient, ctx *ctx.Context, metaSet *metas.Set, idents *idents.Set, tc *memsto.TargetCacheType, uc *memsto.UserCacheType, ugc *memsto.UserGroupCacheType) *Router {
 	return &Router{
 		HTTP:              httpConfig,
 		Center:            center,
 		Alert:             alert,
+		Ibex:              ibex,
 		Operations:        operations,
 		DatasourceCache:   ds,
 		NotifyConfigCache: ncc,
@@ -68,6 +72,7 @@ func New(httpConfig httpx.Config, center cconf.Center, alert aconf.Alert, operat
 		UserCache:         uc,
 		UserGroupCache:    ugc,
 		Ctx:               ctx,
+		HeartbeatHook:     func(ident string) map[string]interface{} { return nil },
 	}
 }
 

center/router/router_heartbeat.go

@@ -3,19 +3,33 @@ package router
 import (
 	"compress/gzip"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
-	"sort"
 	"strings"
 	"time"
 
+	"github.com/ccfos/nightingale/v6/center/metas"
+	"github.com/ccfos/nightingale/v6/memsto"
 	"github.com/ccfos/nightingale/v6/models"
+	"github.com/ccfos/nightingale/v6/pkg/ctx"
+	"github.com/ccfos/nightingale/v6/pushgw/idents"
 
 	"github.com/gin-gonic/gin"
 	"github.com/toolkits/pkg/ginx"
 	"github.com/toolkits/pkg/logger"
 )
 
+type HeartbeatHookFunc func(ident string) map[string]interface{}
+
 func (rt *Router) heartbeat(c *gin.Context) {
+	req, err := HandleHeartbeat(c, rt.Ctx, rt.Alert.Heartbeat.EngineName, rt.MetaSet, rt.IdentSet, rt.TargetCache)
+	ginx.Dangerous(err)
+
+	m := rt.HeartbeatHook(req.Hostname)
+	ginx.NewRender(c).Data(m, err)
+}
+
+func HandleHeartbeat(c *gin.Context, ctx *ctx.Context, engineName string, metaSet *metas.Set, identSet *idents.Set, targetCache *memsto.TargetCacheType) (models.HostMeta, error) {
 	var bs []byte
 	var err error
 	var r *gzip.Reader
@@ -24,7 +38,7 @@ func (rt *Router) heartbeat(c *gin.Context) {
 		r, err = gzip.NewReader(c.Request.Body)
 		if err != nil {
 			c.String(400, err.Error())
-			return
+			return req, err
 		}
 		defer r.Close()
 		bs, err = ioutil.ReadAll(r)
@@ -32,14 +46,18 @@ func (rt *Router) heartbeat(c *gin.Context) {
 	} else {
 		defer c.Request.Body.Close()
 		bs, err = ioutil.ReadAll(c.Request.Body)
-		ginx.Dangerous(err)
+		if err != nil {
+			return req, err
+		}
 	}
 
 	err = json.Unmarshal(bs, &req)
-	ginx.Dangerous(err)
+	if err != nil {
+		return req, err
+	}
 
 	if req.Hostname == "" {
-		ginx.Dangerous("hostname is required", 400)
+		return req, fmt.Errorf("hostname is required", 400)
 	}
 
 	// maybe from pushgw
@@ -52,54 +70,65 @@ func (rt *Router) heartbeat(c *gin.Context) {
 	}
 
 	if req.EngineName == "" {
-		req.EngineName = rt.Alert.Heartbeat.EngineName
+		req.EngineName = engineName
 	}
 
-	rt.MetaSet.Set(req.Hostname, req)
+	metaSet.Set(req.Hostname, req)
 	var items = make(map[string]struct{})
 	items[req.Hostname] = struct{}{}
-	rt.IdentSet.MSet(items)
+	identSet.MSet(items)
 
-	if target, has := rt.TargetCache.Get(req.Hostname); has && target != nil {
+	if target, has := targetCache.Get(req.Hostname); has && target != nil {
 		gid := ginx.QueryInt64(c, "gid", 0)
 		hostIp := strings.TrimSpace(req.HostIp)
 
-		filed := make(map[string]interface{})
+		field := make(map[string]interface{})
 		if gid != 0 && gid != target.GroupId {
-			filed["group_id"] = gid
+			field["group_id"] = gid
 		}
 
 		if hostIp != "" && hostIp != target.HostIp {
-			filed["host_ip"] = hostIp
+			field["host_ip"] = hostIp
 		}
 
-		if len(req.GlobalLabels) > 0 {
+		tagsMap := target.GetTagsMap()
+		tagNeedUpdate := false
+		for k, v := range req.GlobalLabels {
+			if v == "" {
+				continue
+			}
+
+			if tagv, ok := tagsMap[k]; !ok || tagv != v {
+				tagNeedUpdate = true
+				tagsMap[k] = v
+			}
+		}
+
+		if tagNeedUpdate {
 			lst := []string{}
-			for k, v := range req.GlobalLabels {
-				if v == "" {
-					continue
-				}
+			for k, v := range tagsMap {
 				lst = append(lst, k+"="+v)
 			}
-			sort.Strings(lst)
-			labels := strings.Join(lst, " ")
-			if target.Tags != labels {
-				filed["tags"] = labels
-			}
+			labels := strings.Join(lst, " ") + " "
+			field["tags"] = labels
 		}
 
 		if req.EngineName != "" && req.EngineName != target.EngineName {
-			filed["engine_name"] = req.EngineName
+			field["engine_name"] = req.EngineName
 		}
 
-		if len(filed) > 0 {
-			err := target.UpdateFieldsMap(rt.Ctx, filed)
+		if req.AgentVersion != "" && req.AgentVersion != target.AgentVersion {
+			field["agent_version"] = req.AgentVersion
+		}
+
+		if len(field) > 0 {
+			err := target.UpdateFieldsMap(ctx, field)
 			if err != nil {
 				logger.Errorf("update target fields failed, err: %v", err)
 			}
 		}
-		logger.Debugf("heartbeat field:%+v target: %v", filed, *target)
+		logger.Debugf("heartbeat field:%+v target: %v", field, *target)
 	}
 
-	ginx.NewRender(c).Message(err)
+	return req, nil
 }

center/router/router_mute.go

@@ -95,7 +95,8 @@ func (rt *Router) alertMuteAddByService(c *gin.Context) {
 	var f models.AlertMute
 	ginx.BindJSON(c, &f)
 
-	ginx.NewRender(c).Message(f.Add(rt.Ctx))
+	err := f.Add(rt.Ctx)
+	ginx.NewRender(c).Data(f.Id, err)
 }
 
 func (rt *Router) alertMuteDel(c *gin.Context) {

center/router/router_notify_config.go

@@ -90,7 +90,8 @@ func (rt *Router) notifyChannelPuts(c *gin.Context) {
 	var notifyChannels []models.NotifyChannel
 	ginx.BindJSON(c, &notifyChannels)
 
-	channels := []string{models.Dingtalk, models.Wecom, models.Feishu, models.Mm, models.Telegram, models.Email}
+	channels := []string{models.Dingtalk, models.Wecom, models.Feishu, models.Mm, models.Telegram,
+		models.Email, models.Lark, models.LarkCard}
 
 	m := make(map[string]struct{})
 	for _, v := range notifyChannels {
@@ -126,7 +127,8 @@ func (rt *Router) notifyContactPuts(c *gin.Context) {
 	var notifyContacts []models.NotifyContact
 	ginx.BindJSON(c, &notifyContacts)
 
-	keys := []string{models.DingtalkKey, models.WecomKey, models.FeishuKey, models.MmKey, models.TelegramKey}
+	keys := []string{models.DingtalkKey, models.WecomKey, models.FeishuKey, models.MmKey,
+		models.TelegramKey, models.LarkKey}
 
 	m := make(map[string]struct{})
 	for _, v := range notifyContacts {

center/router/router_target.go

@@ -65,12 +65,17 @@ func (rt *Router) targetGets(c *gin.Context) {
 			bgids = append(bgids, 0)
 		}
 	}
-
-	total, err := models.TargetTotal(rt.Ctx, bgids, dsIds, query, downtime)
+	options := []models.BuildTargetWhereOption{
+		models.BuildTargetWhereWithBgids(bgids),
+		models.BuildTargetWhereWithDsIds(dsIds),
+		models.BuildTargetWhereWithQuery(query),
+		models.BuildTargetWhereWithDowntime(downtime),
+	}
+	total, err := models.TargetTotal(rt.Ctx, options...)
 	ginx.Dangerous(err)
 
-	list, err := models.TargetGets(rt.Ctx, bgids, dsIds, query, downtime, limit,
-		ginx.Offset(c, limit), order, desc)
+	list, err := models.TargetGets(rt.Ctx, limit,
+		ginx.Offset(c, limit), order, desc, options...)
 	ginx.Dangerous(err)
 
 	if err == nil {

center/router/router_task.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/toolkits/pkg/ginx"
+	"github.com/toolkits/pkg/i18n"
 	"github.com/toolkits/pkg/str"
 )
 
@@ -104,6 +105,11 @@ func (rt *Router) taskRecordAdd(c *gin.Context) {
 }
 
 func (rt *Router) taskAdd(c *gin.Context) {
+	if !rt.Ibex.Enable {
+		ginx.Bomb(400, i18n.Sprintf(c.GetHeader("X-Language"), "This functionality has not been enabled. Please contact the system administrator to activate it."))
+		return
+	}
+
 	var f models.TaskForm
 	ginx.BindJSON(c, &f)
 

center/router/router_task_tpl.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/toolkits/pkg/ginx"
+	"github.com/toolkits/pkg/i18n"
 	"github.com/toolkits/pkg/str"
 )
 
@@ -118,6 +119,11 @@ type taskTplForm struct {
 }
 
 func (rt *Router) taskTplAdd(c *gin.Context) {
+	if !rt.Ibex.Enable {
+		ginx.Bomb(400, i18n.Sprintf(c.GetHeader("X-Language"), "This functionality has not been enabled. Please contact the system administrator to activate it."))
+		return
+	}
+
 	var f taskTplForm
 	ginx.BindJSON(c, &f)
 

models/notify_config.go

@@ -7,7 +7,11 @@ const NOTIFYCONTACT = "notify_contact"
 const SMTP = "smtp_config"
 const IBEX = "ibex_server"
 
+var GlobalCallback = 0
+var RuleCallback = 1
+
 type Webhook struct {
+	Type          int               `json:"type"`
 	Enable        bool              `json:"enable"`
 	Url           string            `json:"url"`
 	BasicAuthUser string            `json:"basic_auth_user"`
@@ -17,6 +21,9 @@ type Webhook struct {
 	Headers       []string          `json:"headers_str"`
 	SkipVerify    bool              `json:"skip_verify"`
 	Note          string            `json:"note"`
+	RetryCount    int               `json:"retry_count"`
+	RetryInterval int               `json:"retry_interval"`
+	Batch         int               `json:"batch"`
 }
 
 type NotifyScript struct {

models/notify_tpl.go

@@ -171,8 +171,7 @@ func InitNotifyConfig(c *ctx.Context, tplDir string) {
 
 	if cval == "" {
 		var notifyContacts []NotifyContact
-		contacts := []string{DingtalkKey, WecomKey, FeishuKey, MmKey, TelegramKey}
-		for _, contact := range contacts {
+		for _, contact := range DefaultContacts {
 			notifyContacts = append(notifyContacts, NotifyContact{Ident: contact, Name: contact, BuiltIn: true})
 		}
 
@@ -182,6 +181,35 @@ func InitNotifyConfig(c *ctx.Context, tplDir string) {
 			logger.Errorf("failed to set notify contact config: %v", err)
 			return
 		}
+	} else {
+		var contacts []NotifyContact
+		if err = json.Unmarshal([]byte(cval), &contacts); err != nil {
+			logger.Errorf("failed to unmarshal notify channel config: %v", err)
+			return
+		}
+		contactMap := make(map[string]struct{})
+		for _, contact := range contacts {
+			contactMap[contact.Ident] = struct{}{}
+		}
+
+		var newContacts []NotifyContact
+		for _, contact := range DefaultContacts {
+			if _, ok := contactMap[contact]; !ok {
+				newContacts = append(newContacts, NotifyContact{Ident: contact, Name: contact, BuiltIn: true})
+			}
+		}
+		if len(newContacts) > 0 {
+			contacts = append(contacts, newContacts...)
+			data, err := json.Marshal(contacts)
+			if err != nil {
+				logger.Errorf("failed to marshal contacts: %v", err)
+				return
+			}
+			if err = ConfigsSet(c, NOTIFYCONTACT, string(data)); err != nil {
+				logger.Errorf("failed to set notify contact config: %v", err)
+				return
+			}
+		}
 	}
 
 	// init notify tpl
@@ -254,7 +282,8 @@ var TplMap = map[string]string{
 	- {{$key}}: {{$val}}
 {{- end}}
 {{- end}}
-	`,
+{{$domain := "http://请联系管理员修改通知模板将域名替换为实际的域名" }}   
+[事件详情]({{$domain}}/alert-his-events/{{.Id}})|[屏蔽1小时]({{$domain}}/alert-mutes/add?busiGroup={{.GroupId}}&cate={{.Cate}}&datasource_ids={{.DatasourceId}}&prod={{.RuleProd}}{{range $key, $value := .TagsMap}}&tags={{$key}}%3D{{$value}}{{end}})|[查看曲线]({{$domain}}/metric/explorer?data_source_id={{.DatasourceId}}&data_source_name=prometheus&mode=graph&prom_ql={{.PromQl}})`,
 	Email: `<!DOCTYPE html>
 	<html lang="en">
 	<head>
@@ -478,7 +507,10 @@ var TplMap = map[string]string{
 监控指标: {{.TagsJSON}}
 {{if .IsRecovered}}恢复时间：{{timeformat .LastEvalTime}}{{else}}触发时间: {{timeformat .TriggerTime}}
 触发时值: {{.TriggerValue}}{{end}}
-发送时间: {{timestamp}}`,
+发送时间: {{timestamp}}
+{{$domain := "http://请联系管理员修改通知模板将域名替换为实际的域名" }}   
+事件详情: {{$domain}}/alert-his-events/{{.Id}}
+屏蔽1小时: {{$domain}}/alert-mutes/add?busiGroup={{.GroupId}}&cate={{.Cate}}&datasource_ids={{.DatasourceId}}&prod={{.RuleProd}}{{range $key, $value := .TagsMap}}&tags={{$key}}%3D{{$value}}{{end}}`,
 	FeishuCard: `{{ if .IsRecovered }}
 {{- if ne .Cate "host"}}
 **告警集群:** {{.Cluster}}{{end}}   
@@ -495,7 +527,9 @@ var TplMap = map[string]string{
 **发送时间:** {{timestamp}}   
 **触发时值:** {{.TriggerValue}}   
 {{if .RuleNote }}**告警描述:** **{{.RuleNote}}**{{end}}   
-{{- end -}}`,
+{{- end -}}
+{{$domain := "http://请联系管理员修改通知模板将域名替换为实际的域名" }}   
+[事件详情]({{$domain}}/alert-his-events/{{.Id}})|[屏蔽1小时]({{$domain}}/alert-mutes/add?busiGroup={{.GroupId}}&cate={{.Cate}}&datasource_ids={{.DatasourceId}}&prod={{.RuleProd}}{{range $key, $value := .TagsMap}}&tags={{$key}}%3D{{$value}}{{end}})|[查看曲线]({{$domain}}/metric/explorer?data_source_id={{.DatasourceId}}&data_source_name=prometheus&mode=graph&prom_ql={{.PromQl}})`,
 	EmailSubject: `{{if .IsRecovered}}Recovered{{else}}Triggered{{end}}: {{.RuleName}} {{.TagsJSON}}`,
 	Mm: `级别状态: S{{.Severity}} {{if .IsRecovered}}Recovered{{else}}Triggered{{end}}   
 规则名称: {{.RuleName}}{{if .RuleNote}}   
@@ -521,5 +555,38 @@ var TplMap = map[string]string{
 **触发时值**: {{.TriggerValue}}{{end}}   
 {{if .IsRecovered}}**恢复时间**: {{timeformat .LastEvalTime}}{{else}}**首次触发时间**: {{timeformat .FirstTriggerTime}}{{end}}   
 {{$time_duration := sub now.Unix .FirstTriggerTime }}{{if .IsRecovered}}{{$time_duration = sub .LastEvalTime .FirstTriggerTime }}{{end}}**距离首次告警**: {{humanizeDurationInterface $time_duration}}
-**发送时间**: {{timestamp}}`,
+**发送时间**: {{timestamp}}
+{{$domain := "http://请联系管理员修改通知模板将域名替换为实际的域名" }}   
+[事件详情]({{$domain}}/alert-his-events/{{.Id}})|[屏蔽1小时]({{$domain}}/alert-mutes/add?busiGroup={{.GroupId}}&cate={{.Cate}}&datasource_ids={{.DatasourceId}}&prod={{.RuleProd}}{{range $key, $value := .TagsMap}}&tags={{$key}}%3D{{$value}}{{end}})|[查看曲线]({{$domain}}/metric/explorer?data_source_id={{.DatasourceId}}&data_source_name=prometheus&mode=graph&prom_ql={{.PromQl}})`,
+	Lark: `级别状态: S{{.Severity}} {{if .IsRecovered}}Recovered{{else}}Triggered{{end}}   
+规则名称: {{.RuleName}}{{if .RuleNote}}   
+规则备注: {{.RuleNote}}{{end}}   
+监控指标: {{.TagsJSON}}
+{{if .IsRecovered}}恢复时间：{{timeformat .LastEvalTime}}{{else}}触发时间: {{timeformat .TriggerTime}}
+触发时值: {{.TriggerValue}}{{end}}
+发送时间: {{timestamp}}
+{{$domain := "http://请联系管理员修改通知模板将域名替换为实际的域名" }}   
+事件详情: {{$domain}}/alert-his-events/{{.Id}}
+屏蔽1小时: {{$domain}}/alert-mutes/add?busiGroup={{.GroupId}}&cate={{.Cate}}&datasource_ids={{.DatasourceId}}&prod={{.RuleProd}}{{range $key, $value := .TagsMap}}&tags={{$key}}%3D{{$value}}{{end}}`,
+	LarkCard: `{{ if .IsRecovered }}
+{{- if ne .Cate "host"}}
+**告警集群:** {{.Cluster}}{{end}}   
+**级别状态:** S{{.Severity}} Recovered   
+**告警名称:** {{.RuleName}}   
+**恢复时间:** {{timeformat .LastEvalTime}}   
+{{$time_duration := sub now.Unix .FirstTriggerTime }}{{if .IsRecovered}}{{$time_duration = sub .LastEvalTime .FirstTriggerTime }}{{end}}**持续时长**: {{humanizeDurationInterface $time_duration}}   
+**告警描述:** **服务已恢复**   
+{{- else }}
+{{- if ne .Cate "host"}}   
+**告警集群:** {{.Cluster}}{{end}}   
+**级别状态:** S{{.Severity}} Triggered   
+**告警名称:** {{.RuleName}}   
+**触发时间:** {{timeformat .TriggerTime}}   
+**发送时间:** {{timestamp}}   
+**触发时值:** {{.TriggerValue}}
+{{$time_duration := sub now.Unix .FirstTriggerTime }}{{if .IsRecovered}}{{$time_duration = sub .LastEvalTime .FirstTriggerTime }}{{end}}**持续时长**: {{humanizeDurationInterface $time_duration}}   
+{{if .RuleNote }}**告警描述:** **{{.RuleNote}}**{{end}}   
+{{- end -}}
+{{$domain := "http://请联系管理员修改通知模板将域名替换为实际的域名" }}   
+[事件详情]({{$domain}}/alert-his-events/{{.Id}})|[屏蔽1小时]({{$domain}}/alert-mutes/add?busiGroup={{.GroupId}}&cate={{.Cate}}&datasource_ids={{.DatasourceId}}&prod={{.RuleProd}}{{range $key, $value := .TagsMap}}&tags={{$key}}%3D{{$value}}{{end}})|[查看曲线]({{$domain}}/metric/explorer?data_source_id={{.DatasourceId}}&data_source_name=prometheus&mode=graph&prom_ql={{.PromQl}})`,
 }

models/target.go

@@ -85,49 +85,68 @@ func TargetDel(ctx *ctx.Context, idents []string) error {
 	return DB(ctx).Where("ident in ?", idents).Delete(new(Target)).Error
 }
 
-func buildTargetWhere(ctx *ctx.Context, bgids []int64, dsIds []int64, query string, downtime int64) *gorm.DB {
-	session := DB(ctx).Model(&Target{})
+type BuildTargetWhereOption func(session *gorm.DB) *gorm.DB
 
-	if len(bgids) > 0 {
-		session = session.Where("group_id in (?)", bgids)
-	}
-
-	if len(dsIds) > 0 {
-		session = session.Where("datasource_id in (?)", dsIds)
-	}
-
-	if downtime > 0 {
-		session = session.Where("update_at < ?", time.Now().Unix()-downtime)
-	}
-
-	if query != "" {
-		arr := strings.Fields(query)
-		for i := 0; i < len(arr); i++ {
-			q := "%" + arr[i] + "%"
-			session = session.Where("ident like ? or note like ? or tags like ?", q, q, q)
+func BuildTargetWhereWithBgids(bgids []int64) BuildTargetWhereOption {
+	return func(session *gorm.DB) *gorm.DB {
+		if len(bgids) > 0 {
+			session = session.Where("group_id in (?)", bgids)
 		}
+		return session
 	}
+}
 
+func BuildTargetWhereWithDsIds(dsIds []int64) BuildTargetWhereOption {
+	return func(session *gorm.DB) *gorm.DB {
+		if len(dsIds) > 0 {
+			session = session.Where("datasource_id in (?)", dsIds)
+		}
+		return session
+	}
+}
+
+func BuildTargetWhereWithQuery(query string) BuildTargetWhereOption {
+	return func(session *gorm.DB) *gorm.DB {
+		if query != "" {
+			arr := strings.Fields(query)
+			for i := 0; i < len(arr); i++ {
+				q := "%" + arr[i] + "%"
+				session = session.Where("ident like ? or note like ? or tags like ?", q, q, q)
+			}
+		}
+		return session
+	}
+}
+
+func BuildTargetWhereWithDowntime(downtime int64) BuildTargetWhereOption {
+	return func(session *gorm.DB) *gorm.DB {
+		if downtime > 0 {
+			session = session.Where("update_at < ?", time.Now().Unix()-downtime)
+		}
+		return session
+	}
+}
+
+func buildTargetWhere(ctx *ctx.Context, options ...BuildTargetWhereOption) *gorm.DB {
+	session := DB(ctx).Model(&Target{})
+	for _, opt := range options {
+		session = opt(session)
+	}
 	return session
 }
 
-func TargetTotalCount(ctx *ctx.Context) (int64, error) {
-	return Count(DB(ctx).Model(new(Target)))
+func TargetTotal(ctx *ctx.Context, options ...BuildTargetWhereOption) (int64, error) {
+	return Count(buildTargetWhere(ctx, options...))
 }
 
-func TargetTotal(ctx *ctx.Context, bgids []int64, dsIds []int64, query string, downtime int64) (int64, error) {
-	return Count(buildTargetWhere(ctx, bgids, dsIds, query, downtime))
-}
-
-func TargetGets(ctx *ctx.Context, bgids []int64, dsIds []int64, query string, downtime int64, limit, offset int,
-	order string, desc bool) ([]*Target, error) {
+func TargetGets(ctx *ctx.Context, limit, offset int, order string, desc bool, options ...BuildTargetWhereOption) ([]*Target, error) {
 	var lst []*Target
 	if desc {
 		order += " desc"
 	} else {
 		order += " asc"
 	}
-	err := buildTargetWhere(ctx, bgids, dsIds, query, downtime).Order(order).Limit(limit).Offset(offset).Find(&lst).Error
+	err := buildTargetWhere(ctx, options...).Order(order).Limit(limit).Offset(offset).Find(&lst).Error
 	if err == nil {
 		for i := 0; i < len(lst); i++ {
 			lst[i].TagsJSON = strings.Fields(lst[i].Tags)

models/user.go

@@ -30,26 +30,32 @@ const (
 	Telegram     = "telegram"
 	Email        = "email"
 	EmailSubject = "mailsubject"
+	Lark         = "lark"
+	LarkCard     = "larkcard"
 
 	DingtalkKey = "dingtalk_robot_token"
 	WecomKey    = "wecom_robot_token"
 	FeishuKey   = "feishu_robot_token"
 	MmKey       = "mm_webhook_url"
 	TelegramKey = "telegram_robot_token"
+	LarkKey     = "lark_robot_token"
 
 	DingtalkDomain = "oapi.dingtalk.com"
 	WecomDomain    = "qyapi.weixin.qq.com"
 	FeishuDomain   = "open.feishu.cn"
+	LarkDomain     = "open.larksuite.com"
 
 	// FeishuCardDomain The domain name of the feishu card is the same as the feishu,distinguished by the parameter
 	FeishuCardDomain = "open.feishu.cn?card=1"
+	LarkCardDomain   = "open.larksuite.com?card=1"
 	TelegramDomain   = "api.telegram.org"
 	IbexDomain       = "ibex"
 	DefaultDomain    = "default"
 )
 
 var (
-	DefaultChannels = []string{Dingtalk, Wecom, Feishu, Mm, Telegram, Email, FeishuCard}
+	DefaultChannels = []string{Dingtalk, Wecom, Feishu, Mm, Telegram, Email, FeishuCard, Lark, LarkCard}
+	DefaultContacts = []string{DingtalkKey, WecomKey, FeishuKey, MmKey, TelegramKey, LarkKey}
 )
 
 type User struct {
@@ -825,6 +831,9 @@ func (u *User) ExtractToken(key string) (string, bool) {
 		return ret.String(), ret.Exists()
 	case Email:
 		return u.Email, u.Email != ""
+	case Lark, LarkCard:
+		ret := gjson.GetBytes(bs, LarkKey)
+		return ret.String(), ret.Exists()
 	default:
 		return "", false
 	}

pkg/i18nx/var.go

@@ -49,7 +49,8 @@ var I18N = `
       "url path invalid":"url非法",
       "no such server":"无此实例",
       "admin role can not be modified":"管理员角色不允许修改",
-      "builtin payload already exists":"内置模板已存在"
+      "builtin payload already exists":"内置模板已存在",
+      "This functionality has not been enabled. Please contact the system administrator to activate it.":"此功能尚未启用。请联系系统管理员启用"
   },
   "zh_CN": {
     "Username or password invalid": "用户名或密码错误",
@@ -101,7 +102,8 @@ var I18N = `
     "admin role can not be modified":"管理员角色不允许修改",
     "builtin payload already exists":"内置模板已存在",
     "builtin metric already exists":"内置指标已存在",
-    "AlertRule already exists":"告警规则已存在"
+    "AlertRule already exists":"告警规则已存在",
+    "This functionality has not been enabled. Please contact the system administrator to activate it.":"此功能尚未启用。请联系系统管理员启用"
   }
 }
 `

pushgw/router/router.go

@@ -27,6 +27,7 @@ type Router struct {
 	Writers        *writer.WritersType
 	Ctx            *ctx.Context
 	HandleTS       HandleTSFunc
+	HeartbeartApi  string
 }
 
 func New(httpConfig httpx.Config, pushgw pconf.Pushgw, aconf aconf.Alert, tc *memsto.TargetCacheType, bg *memsto.BusiGroupCacheType,

pushgw/router/router_heartbeat.go

@@ -6,14 +6,32 @@ import (
 	"io"
 	"time"
 
+	"github.com/ccfos/nightingale/v6/center/metas"
 	"github.com/ccfos/nightingale/v6/models"
 	"github.com/ccfos/nightingale/v6/pkg/poster"
 	"github.com/gin-gonic/gin"
 	"github.com/toolkits/pkg/ginx"
+	"github.com/toolkits/pkg/logger"
 )
 
 // heartbeat Forward heartbeat request to the center.
 func (rt *Router) heartbeat(c *gin.Context) {
+	gid := ginx.QueryStr(c, "gid", "")
+	req, err := HandleHeartbeat(c, rt.Aconf.Heartbeat.EngineName, rt.MetaSet)
+	if err != nil {
+		logger.Warningf("req:%v heartbeat failed to handle heartbeat err:%v", req, err)
+		ginx.Dangerous(err)
+	}
+	api := "/v1/n9e/heartbeat"
+	if rt.HeartbeartApi != "" {
+		api = rt.HeartbeartApi
+	}
+
+	ret, err := poster.PostByUrlsWithResp[map[string]interface{}](rt.Ctx, api+"?gid="+gid, req)
+	ginx.NewRender(c).Data(ret, err)
+}
+
+func HandleHeartbeat(c *gin.Context, engineName string, metaSet *metas.Set) (models.HostMeta, error) {
 	var bs []byte
 	var err error
 	var r *gzip.Reader
@@ -21,20 +39,26 @@ func (rt *Router) heartbeat(c *gin.Context) {
 	if c.GetHeader("Content-Encoding") == "gzip" {
 		r, err = gzip.NewReader(c.Request.Body)
 		if err != nil {
-			c.String(400, err.Error())
-			return
+			return req, err
 		}
+
 		defer r.Close()
 		bs, err = io.ReadAll(r)
-		ginx.Dangerous(err)
+		if err != nil {
+			return req, err
+		}
 	} else {
 		defer c.Request.Body.Close()
 		bs, err = io.ReadAll(c.Request.Body)
-		ginx.Dangerous(err)
+		if err != nil {
+			return req, err
+		}
 	}
 
 	err = json.Unmarshal(bs, &req)
-	ginx.Dangerous(err)
+	if err != nil {
+		return req, err
+	}
 
 	if req.Hostname == "" {
 		ginx.Dangerous("hostname is required", 400)
@@ -42,11 +66,8 @@ func (rt *Router) heartbeat(c *gin.Context) {
 
 	req.Offset = (time.Now().UnixMilli() - req.UnixTime)
 	req.RemoteAddr = c.ClientIP()
-	gid := ginx.QueryStr(c, "gid", "")
+	req.EngineName = engineName
+	metaSet.Set(req.Hostname, req)
 
-	req.EngineName = rt.Aconf.Heartbeat.EngineName
-
-	rt.MetaSet.Set(req.Hostname, req)
-
-	ginx.NewRender(c).Message(poster.PostByUrls(rt.Ctx, "/v1/n9e/heartbeat?gid="+gid, req))
+	return req, nil
 }