refactor: optimize event processor

Co-authored-by: Haobo Zhang <43698160+haobo8@users.noreply.github.com>

alert/dispatch/dispatch.go

@@ -185,8 +185,8 @@ func (e *Dispatch) HandleEventWithNotifyRule(eventOrigin *models.AlertCurEvent)
 
 			for _, processor := range processors {
 				logger.Infof("before processor notify_id: %d, event:%+v, processor:%+v", notifyRuleId, eventCopy, processor)
-				eventCopy = processor.Process(e.ctx, eventCopy)
-				logger.Infof("after processor notify_id: %d, event:%+v, processor:%+v", notifyRuleId, eventCopy, processor)
+				eventCopy, res, err := processor.Process(e.ctx, eventCopy)
+				logger.Infof("after processor notify_id: %d, event:%+v, processor:%+v, res:%v, err:%v", notifyRuleId, eventCopy, processor, res, err)
 				if eventCopy == nil {
 					logger.Warningf("notify_id: %d, event:%+v, processor:%+v, event is nil", notifyRuleId, eventCopy, processor)
 					break

alert/pipeline/pipeline.go

@@ -1,6 +1,7 @@
 package pipeline
 
 import (
+	_ "github.com/ccfos/nightingale/v6/alert/pipeline/processor/aisummary"
 	_ "github.com/ccfos/nightingale/v6/alert/pipeline/processor/callback"
 	_ "github.com/ccfos/nightingale/v6/alert/pipeline/processor/eventdrop"
 	_ "github.com/ccfos/nightingale/v6/alert/pipeline/processor/eventupdate"

alert/pipeline/processor/aisummary/ai_summary.go

@@ -0,0 +1,198 @@
+package aisummary
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"text/template"
+	"time"
+
+	"github.com/ccfos/nightingale/v6/alert/pipeline/processor/callback"
+	"github.com/ccfos/nightingale/v6/alert/pipeline/processor/common"
+	"github.com/ccfos/nightingale/v6/models"
+	"github.com/ccfos/nightingale/v6/pkg/ctx"
+	"github.com/ccfos/nightingale/v6/pkg/tplx"
+)
+
+const (
+	HTTP_STATUS_SUCCESS_MAX = 299
+)
+
+// AISummaryConfig 配置结构体
+type AISummaryConfig struct {
+	callback.HTTPConfig
+	ModelName      string                 `json:"model_name"`
+	APIKey         string                 `json:"api_key"`
+	PromptTemplate string                 `json:"prompt_template"`
+	CustomParams   map[string]interface{} `json:"custom_params"`
+}
+
+type Message struct {
+	Role    string `json:"role"`
+	Content string `json:"content"`
+}
+
+type ChatCompletionResponse struct {
+	Choices []struct {
+		Message struct {
+			Content string `json:"content"`
+		} `json:"message"`
+	} `json:"choices"`
+}
+
+func init() {
+	models.RegisterProcessor("ai_summary", &AISummaryConfig{})
+}
+
+func (c *AISummaryConfig) Init(settings interface{}) (models.Processor, error) {
+	result, err := common.InitProcessor[*AISummaryConfig](settings)
+	return result, err
+}
+
+func (c *AISummaryConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) (*models.AlertCurEvent, string, error) {
+	if c.Client == nil {
+		if err := c.initHTTPClient(); err != nil {
+			return event, "", fmt.Errorf("failed to initialize HTTP client: %v processor: %v", err, c)
+		}
+	}
+
+	// 准备告警事件信息
+	eventInfo, err := c.prepareEventInfo(event)
+	if err != nil {
+		return event, "", fmt.Errorf("failed to prepare event info: %v processor: %v", err, c)
+	}
+
+	// 调用AI模型生成总结
+	summary, err := c.generateAISummary(eventInfo)
+	if err != nil {
+		return event, "", fmt.Errorf("failed to generate AI summary: %v processor: %v", err, c)
+	}
+
+	// 将总结添加到annotations字段
+	if event.AnnotationsJSON == nil {
+		event.AnnotationsJSON = make(map[string]string)
+	}
+	event.AnnotationsJSON["ai_summary"] = summary
+
+	// 更新Annotations字段
+	b, err := json.Marshal(event.AnnotationsJSON)
+	if err != nil {
+		return event, "", fmt.Errorf("failed to marshal annotations: %v processor: %v", err, c)
+	}
+	event.Annotations = string(b)
+
+	return event, "", nil
+}
+
+func (c *AISummaryConfig) initHTTPClient() error {
+	transport := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: c.SkipSSLVerify},
+	}
+
+	if c.Proxy != "" {
+		proxyURL, err := url.Parse(c.Proxy)
+		if err != nil {
+			return fmt.Errorf("failed to parse proxy url: %v", err)
+		}
+		transport.Proxy = http.ProxyURL(proxyURL)
+	}
+
+	c.Client = &http.Client{
+		Timeout:   time.Duration(c.Timeout) * time.Millisecond,
+		Transport: transport,
+	}
+	return nil
+}
+
+func (c *AISummaryConfig) prepareEventInfo(event *models.AlertCurEvent) (string, error) {
+	var defs = []string{
+		"{{$event := .}}",
+	}
+
+	text := strings.Join(append(defs, c.PromptTemplate), "")
+	t, err := template.New("prompt").Funcs(template.FuncMap(tplx.TemplateFuncMap)).Parse(text)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse prompt template: %v", err)
+	}
+
+	var body bytes.Buffer
+	err = t.Execute(&body, event)
+	if err != nil {
+		return "", fmt.Errorf("failed to execute prompt template: %v", err)
+	}
+
+	return body.String(), nil
+}
+
+func (c *AISummaryConfig) generateAISummary(eventInfo string) (string, error) {
+	// 构建基础请求参数
+	reqParams := map[string]interface{}{
+		"model": c.ModelName,
+		"messages": []Message{
+			{
+				Role:    "user",
+				Content: eventInfo,
+			},
+		},
+	}
+
+	// 合并自定义参数
+	for k, v := range c.CustomParams {
+		reqParams[k] = v
+	}
+
+	// 序列化请求体
+	jsonData, err := json.Marshal(reqParams)
+	if err != nil {
+		return "", fmt.Errorf("failed to marshal request body: %v", err)
+	}
+
+	// 创建HTTP请求
+	req, err := http.NewRequest("POST", c.URL, bytes.NewBuffer(jsonData))
+	if err != nil {
+		return "", fmt.Errorf("failed to create request: %v", err)
+	}
+
+	// 设置请求头
+	req.Header.Set("Authorization", "Bearer "+c.APIKey)
+	req.Header.Set("Content-Type", "application/json")
+	for k, v := range c.Headers {
+		req.Header.Set(k, v)
+	}
+
+	// 发送请求
+	resp, err := c.Client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("failed to send request: %v", err)
+	}
+	defer resp.Body.Close()
+
+	// 检查响应状态码
+	if resp.StatusCode > HTTP_STATUS_SUCCESS_MAX {
+		body, _ := io.ReadAll(resp.Body)
+		return "", fmt.Errorf("unexpected status code: %d, body: %s", resp.StatusCode, string(body))
+	}
+
+	// 读取响应
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("failed to read response body: %v", err)
+	}
+
+	// 解析响应
+	var chatResp ChatCompletionResponse
+	if err := json.Unmarshal(body, &chatResp); err != nil {
+		return "", fmt.Errorf("failed to unmarshal response: %v", err)
+	}
+
+	if len(chatResp.Choices) == 0 {
+		return "", fmt.Errorf("no response from AI model")
+	}
+
+	return chatResp.Choices[0].Message.Content, nil
+}

alert/pipeline/processor/aisummary/ai_summary_test.go

@@ -0,0 +1,69 @@
+package aisummary
+
+import (
+	"testing"
+
+	"github.com/ccfos/nightingale/v6/alert/pipeline/processor/callback"
+	"github.com/ccfos/nightingale/v6/models"
+	"github.com/ccfos/nightingale/v6/pkg/ctx"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAISummaryConfig_Process(t *testing.T) {
+	// 创建测试配置
+	config := &AISummaryConfig{
+		HTTPConfig: callback.HTTPConfig{
+			URL:           "https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
+			Timeout:       30000,
+			SkipSSLVerify: true,
+			Headers: map[string]string{
+				"Content-Type": "application/json",
+			},
+		},
+		ModelName:      "gemini-2.0-flash",
+		APIKey:         "*",
+		PromptTemplate: "告警规则：{{$event.RuleName}}\n严重程度：{{$event.Severity}}",
+		CustomParams: map[string]interface{}{
+			"temperature": 0.7,
+			"max_tokens":  2000,
+			"top_p":       0.9,
+		},
+	}
+
+	// 创建测试事件
+	event := &models.AlertCurEvent{
+		RuleName: "Test Rule",
+		Severity: 1,
+		TagsMap: map[string]string{
+			"host": "test-host",
+		},
+		AnnotationsJSON: map[string]string{
+			"description": "Test alert",
+		},
+	}
+
+	// 测试模板处理
+	eventInfo, err := config.prepareEventInfo(event)
+	assert.NoError(t, err)
+	assert.Contains(t, eventInfo, "Test Rule")
+	assert.Contains(t, eventInfo, "1")
+
+	// 测试配置初始化
+	processor, err := config.Init(config)
+	assert.NoError(t, err)
+	assert.NotNil(t, processor)
+
+	// 测试处理函数
+	result, _, err := processor.Process(&ctx.Context{}, event)
+	assert.NoError(t, err)
+	assert.NotNil(t, result)
+	assert.NotEmpty(t, result.AnnotationsJSON["ai_summary"])
+
+	// 展示处理结果
+	t.Log("\n=== 处理结果 ===")
+	t.Logf("告警规则: %s", result.RuleName)
+	t.Logf("严重程度: %d", result.Severity)
+	t.Logf("标签: %v", result.TagsMap)
+	t.Logf("原始注释: %v", result.AnnotationsJSON["description"])
+	t.Logf("AI总结: %s", result.AnnotationsJSON["ai_summary"])
+}

alert/pipeline/processor/callback/callback.go

@@ -3,6 +3,7 @@ package callback
 import (
 	"crypto/tls"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -42,7 +43,7 @@ func (c *CallbackConfig) Init(settings interface{}) (models.Processor, error) {
 	return result, err
 }
 
-func (c *CallbackConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) *models.AlertCurEvent {
+func (c *CallbackConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) (*models.AlertCurEvent, string, error) {
 	if c.Client == nil {
 		transport := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: c.SkipSSLVerify},
@@ -51,7 +52,7 @@ func (c *CallbackConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent)
 		if c.Proxy != "" {
 			proxyURL, err := url.Parse(c.Proxy)
 			if err != nil {
-				logger.Errorf("failed to parse proxy url: %v", err)
+				return event, "", fmt.Errorf("failed to parse proxy url: %v processor: %v", err, c)
 			} else {
 				transport.Proxy = http.ProxyURL(proxyURL)
 			}
@@ -71,14 +72,12 @@ func (c *CallbackConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent)
 
 	body, err := json.Marshal(event)
 	if err != nil {
-		logger.Errorf("failed to marshal event: %v", err)
-		return event
+		return event, "", fmt.Errorf("failed to marshal event: %v processor: %v", err, c)
 	}
 
 	req, err := http.NewRequest("POST", c.URL, strings.NewReader(string(body)))
 	if err != nil {
-		logger.Errorf("failed to create request: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("failed to create request: %v processor: %v", err, c)
 	}
 
 	for k, v := range headers {
@@ -91,16 +90,14 @@ func (c *CallbackConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent)
 
 	resp, err := c.Client.Do(req)
 	if err != nil {
-		logger.Errorf("failed to send request: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("failed to send request: %v processor: %v", err, c)
 	}
 
 	b, err := io.ReadAll(resp.Body)
 	if err != nil {
-		logger.Errorf("failed to read response body: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("failed to read response body: %v processor: %v", err, c)
 	}
 
-	logger.Infof("response body: %s", string(b))
-	return event
+	logger.Debugf("callback processor response body: %s", string(b))
+	return event, "callback success", nil
 }

alert/pipeline/processor/eventdrop/event_drop.go

@@ -2,6 +2,7 @@ package eventdrop
 
 import (
 	"bytes"
+	"fmt"
 	"strings"
 	texttemplate "text/template"
 
@@ -25,7 +26,7 @@ func (c *EventDropConfig) Init(settings interface{}) (models.Processor, error) {
 	return result, err
 }
 
-func (c *EventDropConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) *models.AlertCurEvent {
+func (c *EventDropConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) (*models.AlertCurEvent, string, error) {
 	// 使用背景是可以根据此处理器，实现对事件进行更加灵活的过滤的逻辑
 	// 在标签过滤和属性过滤都不满足需求时可以使用
 	// 如果模板执行结果为 true，则删除该事件
@@ -40,22 +41,20 @@ func (c *EventDropConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent)
 
 	tpl, err := texttemplate.New("eventdrop").Funcs(tplx.TemplateFuncMap).Parse(text)
 	if err != nil {
-		logger.Errorf("processor failed to parse template: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("processor failed to parse template: %v processor: %v", err, c)
 	}
 
 	var body bytes.Buffer
 	if err = tpl.Execute(&body, event); err != nil {
-		logger.Errorf("processor failed to execute template: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("processor failed to execute template: %v processor: %v", err, c)
 	}
 
 	result := strings.TrimSpace(body.String())
 	logger.Infof("processor eventdrop result: %v", result)
 	if result == "true" {
 		logger.Infof("processor eventdrop drop event: %v", event)
-		return nil
+		return event, "drop event success", nil
 	}
 
-	return event
+	return event, "drop event failed", nil
 }

alert/pipeline/processor/eventupdate/event_update.go

@@ -3,6 +3,7 @@ package eventupdate
 import (
 	"crypto/tls"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -30,7 +31,7 @@ func (c *EventUpdateConfig) Init(settings interface{}) (models.Processor, error)
 	return result, err
 }
 
-func (c *EventUpdateConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) *models.AlertCurEvent {
+func (c *EventUpdateConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) (*models.AlertCurEvent, string, error) {
 	if c.Client == nil {
 		transport := &http.Transport{
 			TLSClientConfig: &tls.Config{InsecureSkipVerify: c.SkipSSLVerify},
@@ -39,7 +40,7 @@ func (c *EventUpdateConfig) Process(ctx *ctx.Context, event *models.AlertCurEven
 		if c.Proxy != "" {
 			proxyURL, err := url.Parse(c.Proxy)
 			if err != nil {
-				logger.Errorf("failed to parse proxy url: %v", err)
+				return event, "", fmt.Errorf("failed to parse proxy url: %v processor: %v", err, c)
 			} else {
 				transport.Proxy = http.ProxyURL(proxyURL)
 			}
@@ -59,14 +60,12 @@ func (c *EventUpdateConfig) Process(ctx *ctx.Context, event *models.AlertCurEven
 
 	body, err := json.Marshal(event)
 	if err != nil {
-		logger.Errorf("failed to marshal event: %v", err)
-		return event
+		return event, "", fmt.Errorf("failed to marshal event: %v processor: %v", err, c)
 	}
 
 	req, err := http.NewRequest("POST", c.URL, strings.NewReader(string(body)))
 	if err != nil {
-		logger.Errorf("failed to create request: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("failed to create request: %v processor: %v", err, c)
 	}
 
 	for k, v := range headers {
@@ -79,22 +78,19 @@ func (c *EventUpdateConfig) Process(ctx *ctx.Context, event *models.AlertCurEven
 
 	resp, err := c.Client.Do(req)
 	if err != nil {
-		logger.Errorf("failed to send request: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("failed to send request: %v processor: %v", err, c)
 	}
 
 	b, err := io.ReadAll(resp.Body)
 	if err != nil {
-		logger.Errorf("failed to read response body: %v event: %v", err, event)
-		return event
+		return nil, "", fmt.Errorf("failed to read response body: %v processor: %v", err, c)
 	}
-	logger.Infof("response body: %s", string(b))
+	logger.Debugf("event update processor response body: %s", string(b))
 
 	err = json.Unmarshal(b, &event)
 	if err != nil {
-		logger.Errorf("failed to unmarshal response body: %v event: %v", err, event)
-		return event
+		return event, "", fmt.Errorf("failed to unmarshal response body: %v processor: %v", err, c)
 	}
 
-	return event
+	return event, "", nil
 }

alert/pipeline/processor/relabel/relabel.go

@@ -42,7 +42,7 @@ func (r *RelabelConfig) Init(settings interface{}) (models.Processor, error) {
 	return result, err
 }
 
-func (r *RelabelConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) *models.AlertCurEvent {
+func (r *RelabelConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) (*models.AlertCurEvent, string, error) {
 	sourceLabels := make([]model.LabelName, len(r.SourceLabels))
 	for i := range r.SourceLabels {
 		sourceLabels[i] = model.LabelName(strings.ReplaceAll(r.SourceLabels[i], ".", REPLACE_DOT))
@@ -64,7 +64,7 @@ func (r *RelabelConfig) Process(ctx *ctx.Context, event *models.AlertCurEvent) *
 	}
 
 	EventRelabel(event, relabelConfigs)
-	return event
+	return event, "", nil
 }
 
 func EventRelabel(event *models.AlertCurEvent, relabelConfigs []*pconf.RelabelConfig) {

center/router/router_event_pipeline.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/toolkits/pkg/ginx"
-	"github.com/toolkits/pkg/logger"
 )
 
 // 获取事件Pipeline列表
@@ -145,9 +144,17 @@ func (rt *Router) tryRunEventPipeline(c *gin.Context) {
 		if err != nil {
 			ginx.Bomb(http.StatusBadRequest, "get processor: %+v err: %+v", p, err)
 		}
-		event = processor.Process(rt.Ctx, event)
+		event, _, err = processor.Process(rt.Ctx, event)
+		if err != nil {
+			ginx.Bomb(http.StatusBadRequest, "processor: %+v err: %+v", p, err)
+		}
+
 		if event == nil {
-			ginx.Bomb(http.StatusBadRequest, "event is dropped")
+			ginx.NewRender(c).Data(map[string]interface{}{
+				"event":  event,
+				"result": "event is dropped",
+			}, nil)
+			return
 		}
 	}
 
@@ -172,13 +179,15 @@ func (rt *Router) tryRunEventProcessor(c *gin.Context) {
 	if err != nil {
 		ginx.Bomb(http.StatusBadRequest, "get processor err: %+v", err)
 	}
-	event = processor.Process(rt.Ctx, event)
-	logger.Infof("processor %+v result: %+v", f.ProcessorConfig, event)
-	if event == nil {
-		ginx.Bomb(http.StatusBadRequest, "event is dropped")
+	event, res, err := processor.Process(rt.Ctx, event)
+	if err != nil {
+		ginx.Bomb(http.StatusBadRequest, "processor err: %+v", err)
 	}
 
-	ginx.NewRender(c).Data(event, nil)
+	ginx.NewRender(c).Data(map[string]interface{}{
+		"event":  event,
+		"result": res,
+	}, nil)
 }
 
 func (rt *Router) tryRunEventProcessorByNotifyRule(c *gin.Context) {
@@ -212,9 +221,17 @@ func (rt *Router) tryRunEventProcessorByNotifyRule(c *gin.Context) {
 			if err != nil {
 				ginx.Bomb(http.StatusBadRequest, "get processor: %+v err: %+v", p, err)
 			}
-			event = processor.Process(rt.Ctx, event)
+
+			event, _, err := processor.Process(rt.Ctx, event)
+			if err != nil {
+				ginx.Bomb(http.StatusBadRequest, "processor: %+v err: %+v", p, err)
+			}
 			if event == nil {
-				ginx.Bomb(http.StatusBadRequest, "event is dropped")
+				ginx.NewRender(c).Data(map[string]interface{}{
+					"event":  event,
+					"result": "event is dropped",
+				}, nil)
+				return
 			}
 		}
 	}

models/event_processor.go

@@ -8,8 +8,12 @@ import (
 )
 
 type Processor interface {
-	Init(settings interface{}) (Processor, error)                  // 初始化配置
-	Process(ctx *ctx.Context, event *AlertCurEvent) *AlertCurEvent // 处理告警事件
+	Init(settings interface{}) (Processor, error) // 初始化配置
+	Process(ctx *ctx.Context, event *AlertCurEvent) (*AlertCurEvent, string, error)
+	// 处理器有三种情况：
+	// 1. 处理成功，返回处理后的事件
+	// 2. 处理成功，不需要返回处理后端事件，只返回处理结果，将处理结果放到 string 中，比如 eventdrop callback 处理器
+	// 3. 处理失败，返回错误，将错误放到 error 中
 }
 
 type NewProcessorFn func(settings interface{}) (Processor, error)