From d53e1a174480c104f9072dd26c7d0d971d4ac7df Mon Sep 17 00:00:00 2001 From: liangyanbin Date: Mon, 26 Oct 2026 10:36:47 +0800 Subject: [PATCH 1/3] feat(alert): attach screenshots to FlashDuty alerts --- alert/sender/provider/flashduty_provider.go | 143 ++++++- .../provider/flashduty_provider_test.go | 360 ++++++++++++++++++ 2 files changed, 502 insertions(+), 1 deletion(-) create mode 100644 alert/sender/provider/flashduty_provider_test.go diff --git a/alert/sender/provider/flashduty_provider.go b/alert/sender/provider/flashduty_provider.go index 14adfebec..34fc889d0 100644 --- a/alert/sender/provider/flashduty_provider.go +++ b/alert/sender/provider/flashduty_provider.go @@ -7,7 +7,10 @@ import ( "errors" "fmt" "io" + "mime/multipart" "net/http" + "net/url" + "sort" "strconv" "strings" "time" @@ -35,6 +38,10 @@ func (p *FlashDutyProvider) Notify(ctx context.Context, req *NotifyRequest) *Not if req.Config.RequestConfig == nil || req.Config.RequestConfig.FlashDutyRequestConfig == nil { return &NotifyResult{Target: "", Response: "", Err: errors.New("flashduty request config not found")} } + events, err := prepareFlashDutyEvents(req.Config.RequestConfig.FlashDutyRequestConfig, req.Events, req.HttpClient) + if err != nil { + return &NotifyResult{Err: err} + } channelIDs := req.FlashDutyChannelIDs if len(channelIDs) == 0 { channelIDs = []int64{0} @@ -45,7 +52,7 @@ func (p *FlashDutyProvider) Notify(ctx context.Context, req *NotifyRequest) *Not for _, channelID := range channelIDs { target := strconv.FormatInt(channelID, 10) targets = append(targets, target) - resp, err := SendFlashDuty(req.Config.RequestConfig.FlashDutyRequestConfig, req.Events, channelID, req.HttpClient) + resp, err := sendFlashDutyEvents(req.Config.RequestConfig.FlashDutyRequestConfig, events, channelID, req.HttpClient) if err != nil { // 单个 channel 失败不要中断其他 channel 的发送 failedMsgs = append(failedMsgs, fmt.Sprintf("channel %s: %v", target, err)) @@ -62,6 +69,140 @@ func (p *FlashDutyProvider) Notify(ctx context.Context, req *NotifyRequest) *Not } func SendFlashDuty(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, flashDutyChannelID int64, client *http.Client) (string, error) { + events, err := prepareFlashDutyEvents(config, events, client) + if err != nil { + return "", err + } + return sendFlashDutyEvents(config, events, flashDutyChannelID, client) +} + +// Prepare private event copies so other notification channels can still use the screenshots. +func prepareFlashDutyEvents(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, client *http.Client) ([]*models.AlertCurEvent, error) { + if client == nil { + return nil, errors.New("http client not found") + } + prepared := make([]*models.AlertCurEvent, len(events)) + for i, event := range events { + if event == nil { + continue + } + copied := *event + copied.ShotImageBase64 = nil + prepared[i] = &copied + // The snapshot field holds one image. Pick it consistently when multiple screenshots exist. + keys := make([]string, 0, len(event.ShotImageBase64)) + for key := range event.ShotImageBase64 { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + imageBase64 := event.ShotImageBase64[key] + if strings.TrimSpace(imageBase64) == "" { + continue + } + var ruleConfig map[string]json.RawMessage + configJSON := []byte(event.RuleConfig) + if event.RuleConfigJson != nil { + var err error + configJSON, err = json.Marshal(event.RuleConfigJson) + if err != nil { + logger.Warningf("send_flashduty: skip image attachment event_id=%d: marshal rule config: %v", event.Id, err) + break + } + } + if len(configJSON) > 0 { + if err := json.Unmarshal(configJSON, &ruleConfig); err != nil { + logger.Warningf("send_flashduty: skip image attachment event_id=%d: parse rule config: %v", event.Id, err) + break + } + } + if ruleConfig == nil { + ruleConfig = make(map[string]json.RawMessage) + } + imageKey, err := uploadFlashDutyImage(config.IntegrationUrl, imageBase64, client) + if err != nil { + // Screenshots are optional: keep sending alerts even if uploading fails. + logger.Warningf("send_flashduty: upload image failed event_id=%d: %v", event.Id, err) + break + } + // Match NewNorthstarAlertEvent: annotations.img_key and a string snapshot_pic_url. + ruleConfig["snapshot_pic_url"], _ = json.Marshal(imageKey) + copied.RuleConfigJson = ruleConfig + copied.AnnotationsJSON = make(map[string]string, len(event.AnnotationsJSON)+1) + for name, value := range event.AnnotationsJSON { + copied.AnnotationsJSON[name] = value + } + copied.AnnotationsJSON["img_key"] = imageKey + break + } + } + return prepared, nil +} + +func uploadFlashDutyImage(integrationURL, imageBase64 string, client *http.Client) (string, error) { + uploadURL, err := url.Parse(integrationURL) + if err != nil { + return "", err + } + query := url.Values{"integration_key": {uploadURL.Query().Get("integration_key")}} + uploadURL.Path = "/push/image/upload" + uploadURL.RawPath = "" + uploadURL.RawQuery = query.Encode() + uploadURL.Fragment = "" + imgBytes, err := decodeBase64Payload(imageBase64) + if err != nil { + return "", fmt.Errorf("decode image: %w", err) + } + if len(imgBytes) == 0 || len(imgBytes) > 5*1024*1024 { + return "", errors.New("image must be non-empty and no larger than 5MB") + } + var body bytes.Buffer + writer := multipart.NewWriter(&body) + part, err := writer.CreateFormFile("image", "snapshot.png") + if err != nil { + return "", err + } + if _, err = part.Write(imgBytes); err != nil { + return "", err + } + if err = writer.Close(); err != nil { + return "", err + } + req, err := http.NewRequest(http.MethodPost, uploadURL.String(), &body) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", writer.FormDataContentType()) + resp, err := client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "", fmt.Errorf("upload image returned status %d", resp.StatusCode) + } + var result struct { + Data struct { + ImageKey string `json:"image_key"` + } `json:"data"` + Error *struct { + Code string `json:"code"` + Message string `json:"message"` + } `json:"error"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", fmt.Errorf("parse upload image response: %w", err) + } + if result.Error != nil { + return "", fmt.Errorf("upload image failed: %s: %s", result.Error.Code, result.Error.Message) + } + if result.Data.ImageKey == "" { + return "", errors.New("upload image response has no image_key") + } + return result.Data.ImageKey, nil +} + +func sendFlashDutyEvents(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, flashDutyChannelID int64, client *http.Client) (string, error) { // todo 每一个 channel 批量发送事件 if client == nil { return "", fmt.Errorf("http client not found") diff --git a/alert/sender/provider/flashduty_provider_test.go b/alert/sender/provider/flashduty_provider_test.go new file mode 100644 index 000000000..e4e8b31f7 --- /dev/null +++ b/alert/sender/provider/flashduty_provider_test.go @@ -0,0 +1,360 @@ +package provider + +import ( + "os" + "context" + "encoding/base64" + "encoding/json" + "io" + + "time" + + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + "github.com/ccfos/nightingale/v6/models" +) + +func TestFlashDutyUploadBeforeNotify(t *testing.T) { + uploads, sends := 0, 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Query().Get("integration_key") != "test/key" { + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + } + if r.URL.Path == "/push/image/upload" { + uploads++ + if len(r.URL.Query()) != 1 { + t.Errorf("unexpected upload query: %s", r.URL.RawQuery) + } + f, _, err := r.FormFile("image") + if err != nil { + t.Error(err) + w.WriteHeader(400) + return + } + defer f.Close() + b, _ := io.ReadAll(f) + if string(b) != "image bytes" { + t.Errorf("unexpected image: %q", b) + } + io.WriteString(w, `{"data":{"image_key":"img_test"}}`) + return + } + sends++ + if uploads != 1 { + t.Errorf("uploads before send: %d", uploads) + } + if r.URL.Path != "/event/push" || r.URL.Query().Get("channel_id") == "" { + t.Errorf("unexpected event URL: %s", r.URL) + } + var events []struct { + RuleConfig struct { + Snapshot string `json:"snapshot_pic_url"` + Existing string `json:"existing"` + } `json:"rule_config"` + Shot map[string]string `json:"shot_image_base64"` + Annotations map[string]string `json:"annotations"` + } + if err := json.NewDecoder(r.Body).Decode(&events); err != nil { + t.Error(err) + return + } + if len(events) != 1 { + t.Errorf("events: %v", events) + return + } + if events[0].Annotations["img_key"] != "img_test" || events[0].Shot != nil || events[0].RuleConfig.Snapshot != "img_test" || events[0].RuleConfig.Existing != "preserved" { + t.Errorf("unexpected payload: %+v", events[0]) + } + io.WriteString(w, `{}`) + })) + defer server.Close() + original := &models.AlertCurEvent{ + RuleConfigJson: map[string]interface{}{"existing": "preserved"}, + ShotImageBase64: map[string]string{"panel": "data:image/png;base64,aW1hZ2UgYnl0ZXM="}, + } + result := (&FlashDutyProvider{}).Notify(context.Background(), &NotifyRequest{ + Config: &models.NotifyChannelConfig{RequestConfig: &models.RequestConfig{FlashDutyRequestConfig: &models.FlashDutyRequestConfig{IntegrationUrl: server.URL + "/event/push?integration_key=test%2Fkey"}}}, + Events: []*models.AlertCurEvent{original}, HttpClient: server.Client(), FlashDutyChannelIDs: []int64{1, 2}, + }) + if result.Err != nil { + t.Fatal(result.Err) + } + if uploads != 1 || sends != 2 { + t.Fatalf("uploads=%d sends=%d", uploads, sends) + } + if len(original.ShotImageBase64) != 1 || len(original.RuleConfigJson.(map[string]interface{})) != 1 { + t.Fatal("original event was changed") + } +} + +func TestFlashDutyImagePreparation(t *testing.T) { + for _, tc := range []struct { + name, response, image string + status int + uploadFails bool + }{ + {"raw base64", `{"data":{"image_key":"img_test"}}`, "aW1hZ2U=", 200, false}, + {"HTTP error", `{}`, "aW1hZ2U=", 429, true}, + {"API error", `{"error":{"code":"InvalidParameter"}}`, "aW1hZ2U=", 200, true}, + {"missing key", `{"data":{}}`, "aW1hZ2U=", 200, true}, + {"invalid JSON", `{`, "aW1hZ2U=", 200, true}, + {"invalid base64", `{}`, "!!!", 200, true}, + } { + t.Run(tc.name, func(t *testing.T) { + sends := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/push/image/upload" { + sends++ + var sent []models.AlertCurEvent + if err := json.NewDecoder(r.Body).Decode(&sent); err != nil { + t.Error(err) + return + } + if len(sent) != 2 || sent[0].ShotImageBase64 != nil || sent[1].Id != 2 { + t.Errorf("unexpected sent events: %+v", sent) + } + io.WriteString(w, `{}`) + return + } + w.WriteHeader(tc.status) + io.WriteString(w, tc.response) + })) + defer server.Close() + event := &models.AlertCurEvent{RuleConfig: `{"existing":1234567890123456789}`, ShotImageBase64: map[string]string{"panel": tc.image}} + config := &models.FlashDutyRequestConfig{IntegrationUrl: server.URL + "/push?integration_key=test"} + prepared, err := prepareFlashDutyEvents(config, []*models.AlertCurEvent{event}, server.Client()) + if err != nil { + t.Fatalf("error=%v", err) + } + if !tc.uploadFails { + b, _ := json.Marshal(prepared[0]) + if !strings.Contains(string(b), `"existing":1234567890123456789`) || !strings.Contains(string(b), `"snapshot_pic_url":"img_test"`) { + t.Fatalf("unexpected payload %s", b) + } + } else if prepared[0].RuleConfigJson != nil { + t.Fatal("failed upload changed rule config") + } + if prepared[0].ShotImageBase64 != nil || len(event.ShotImageBase64) != 1 { + t.Fatal("screenshot was not stripped from copy or original was changed") + } + if _, err := SendFlashDuty(config, []*models.AlertCurEvent{event, {Id: 2}}, 0, server.Client()); err != nil { + t.Fatalf("image upload must not block alerts: %v", err) + } + if sends != 1 { + t.Fatalf("expected alert batch to be sent, got %d sends", sends) + } + }) + } + event := &models.AlertCurEvent{RuleConfigJson: map[string]interface{}{"existing": "value"}} + prepared, err := prepareFlashDutyEvents(nil, []*models.AlertCurEvent{nil, event}, http.DefaultClient) + if err != nil || prepared[0] != nil || prepared[1].RuleConfigJson == nil { + t.Fatalf("no-image preparation failed: %v", err) + } +} + +// Exercise transport failures without relying on timing or external endpoints. +type flashDutyTestTransport func(*http.Request) (*http.Response, error) + +func (f flashDutyTestTransport) RoundTrip(r *http.Request) (*http.Response, error) { + return f(r) +} + +type flashDutyFailedBody struct{} + +func (flashDutyFailedBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF } +func (flashDutyFailedBody) Close() error { return nil } + +func TestFlashDutyOptionalImageFailures(t *testing.T) { + for _, failure := range []string{"connection", "response read", "non-object config", "malformed stored config"} { + t.Run(failure, func(t *testing.T) { + uploads, attempts := 0, 0 + channels := make(map[string]int) + original := &models.AlertCurEvent{ + Id: 1, + RuleConfigJson: map[string]interface{}{"existing": "preserved"}, + ShotImageBase64: map[string]string{"panel": "aW1hZ2U="}, + } + if failure == "non-object config" { + original.RuleConfigJson = []string{"preserved"} + } + if failure == "malformed stored config" { + original.RuleConfigJson = nil + original.RuleConfig = "{" + } + originalConfig, _ := json.Marshal(original.RuleConfigJson) + client := &http.Client{Transport: flashDutyTestTransport(func(r *http.Request) (*http.Response, error) { + if r.URL.Path == "/push/image/upload" { + uploads++ + if failure == "connection" { + return nil, io.ErrUnexpectedEOF + } + return &http.Response{StatusCode: 200, Body: flashDutyFailedBody{}, Header: make(http.Header)}, nil + } + attempts++ + if attempts == 1 { + return nil, io.ErrUnexpectedEOF // Existing event retry must still work. + } + var sent []models.AlertCurEvent + if err := json.NewDecoder(r.Body).Decode(&sent); err != nil { + t.Fatal(err) + } + if len(sent) != 2 || sent[0].ShotImageBase64 != nil || sent[1].Id != 2 { + t.Fatalf("unexpected batch: %+v", sent) + } + config, _ := json.Marshal(sent[0].RuleConfigJson) + if string(config) != string(originalConfig) { + t.Errorf("rule config changed: %s -> %s", originalConfig, config) + } + channels[r.URL.Query().Get("channel_id")]++ + return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{}`)), Header: make(http.Header)}, nil + })} + result := (&FlashDutyProvider{}).Notify(context.Background(), &NotifyRequest{ + Config: &models.NotifyChannelConfig{RequestConfig: &models.RequestConfig{FlashDutyRequestConfig: &models.FlashDutyRequestConfig{ + IntegrationUrl: "http://flashduty.test/event/push?integration_key=test", RetryTimes: 1, RetrySleep: 1, + }}}, + Events: []*models.AlertCurEvent{original, {Id: 2}}, HttpClient: client, FlashDutyChannelIDs: []int64{1, 2}, + }) + if result.Err != nil || attempts != 3 || channels["1"] != 1 || channels["2"] != 1 { + t.Fatalf("result=%+v attempts=%d channels=%v", result, attempts, channels) + } + wantUploads := 1 + if failure == "non-object config" || failure == "malformed stored config" { + wantUploads = 0 + } + if uploads != wantUploads || len(original.ShotImageBase64) != 1 { + t.Fatalf("uploads=%d; original screenshots=%v", uploads, original.ShotImageBase64) + } + }) + } +} + +func TestFlashDutyPreservesEventData(t *testing.T) { + for _, tc := range []struct { + name, ruleConfig string + uploadSucceeds bool + }{ + {"add snapshot", `{"queries":[{"limit":1234567890123456789}],"name":"rule"}`, true}, + {"replace snapshot object", `{"snapshot_pic_url":{"url":"https://example.com/image.png","extra":{"value":1234567890123456789}},"name":"rule"}`, true}, + {"existing image key", `{"snapshot_pic_url":{"img_key":"img_existing","url":"keep"}}`, true}, + {"existing string snapshot", `{"snapshot_pic_url":"https://example.com/image.png"}`, true}, + {"existing annotation key", `{"name":"rule"}`, true}, + {"empty snapshot", `{"snapshot_pic_url":""}`, true}, + {"null snapshot", `{"snapshot_pic_url":null}`, true}, + {"upload failure preserves old keys", `{"snapshot_pic_url":"img_existing"}`, false}, + } { + t.Run(tc.name, func(t *testing.T) { + original := &models.AlertCurEvent{ + Id: 42, RuleName: "rule", RuleConfig: tc.ruleConfig, + RuleConfigJson: json.RawMessage(tc.ruleConfig), + AnnotationsJSON: map[string]string{"note": "kept"}, + TagsJSON: []string{"env=prod"}, ShotImageBase64: map[string]string{"panel": "aW1hZ2U="}, + } + if tc.name == "existing annotation key" || !tc.uploadSucceeds { + original.AnnotationsJSON["img_key"] = "img_existing" + } + before, _ := json.Marshal(original) + uploads := 0 + client := &http.Client{Transport: flashDutyTestTransport(func(r *http.Request) (*http.Response, error) { + uploads++ + if !tc.uploadSucceeds { + return nil, io.ErrUnexpectedEOF + } + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"data":{"image_key":"img_new"}}`))}, nil + })} + prepared, err := prepareFlashDutyEvents(&models.FlashDutyRequestConfig{IntegrationUrl: "http://flashduty.test/push?integration_key=test"}, []*models.AlertCurEvent{original}, client) + if err != nil { + t.Fatal(err) + } + after, _ := json.Marshal(original) + if string(before) != string(after) { + t.Fatal("original event changed") + } + if prepared[0].ShotImageBase64 != nil { + t.Fatal("outgoing screenshot not cleared") + } + if uploads != 1 { + t.Fatalf("uploads=%d", uploads) + } + out, _ := json.Marshal(prepared[0]) + var got, want map[string]json.RawMessage + json.Unmarshal(out, &got) + json.Unmarshal(before, &want) + // Remove only the intended changes, then compare all original event data. + delete(got, "shot_image_base64") + delete(want, "shot_image_base64") + if tc.uploadSucceeds { + var config map[string]json.RawMessage + json.Unmarshal(got["rule_config"], &config) + if string(config["snapshot_pic_url"]) != `"img_new"` { + t.Fatalf("unexpected snapshot: %s", config["snapshot_pic_url"]) + } + var originalConfig map[string]json.RawMessage + json.Unmarshal(want["rule_config"], &originalConfig) + if old, exists := originalConfig["snapshot_pic_url"]; exists { + config["snapshot_pic_url"] = old + } else { + delete(config, "snapshot_pic_url") + } + var annotations map[string]string + json.Unmarshal(got["annotations"], &annotations) + if annotations["img_key"] != "img_new" { + t.Fatalf("unexpected annotations: %v", annotations) + } + if old, exists := original.AnnotationsJSON["img_key"]; exists { + annotations["img_key"] = old + } else { + delete(annotations, "img_key") + } + got["annotations"], _ = json.Marshal(annotations) + got["rule_config"], _ = json.Marshal(config) + } + // Decode using json.Number to retain integer precision while ignoring object key order. + decode := func(v interface{}) interface{} { + b, _ := json.Marshal(v) + d := json.NewDecoder(strings.NewReader(string(b))) + d.UseNumber() + var result interface{} + if err := d.Decode(&result); err != nil { + t.Fatal(err) + } + return result + } + if !reflect.DeepEqual(decode(got), decode(want)) { + t.Fatalf("existing event data changed: %s", out) + } + }) + } +} + +var testEvents = []*models.AlertCurEvent{} + +func TestFlashDutySend(t *testing.T) { + + for _, event := range testEvents { + // 读取 ～/workspace/tmp/3.png + imageData, err := os.ReadFile("/Users/liangyb/workspace/tmp/3.png") + if err != nil { + t.Fatal(err) + } + event.ShotImageBase64 = map[string]string{"panel": base64.StdEncoding.EncodeToString(imageData)} + } + + // 创建NotifyChannelConfig对象 + config := &models.RequestConfig{ + FlashDutyRequestConfig: &models.FlashDutyRequestConfig{ + IntegrationUrl: "https://api.flashcat.cloud/event/push/alert/n9e?integration_key=xxx", + }, + } + + httpClient := &http.Client{ + Timeout: time.Duration(10) * time.Second, + } + + flashDutyChannelID := int64(4344322009498) + resp, err := SendFlashDuty(config.FlashDutyRequestConfig, testEvents, flashDutyChannelID, httpClient) + t.Logf("flashduty resp=%q err=%v", resp, err) +} From 390e954dd0ccac15ac1b4ead66628670c78d89d8 Mon Sep 17 00:00:00 2001 From: liangyanbin Date: Mon, 26 Oct 2026 13:06:59 +0800 Subject: [PATCH 2/3] refactor(alert): upload FlashDuty images once per channel instead of once overall --- alert/sender/provider/flashduty_provider.go | 22 +++++-------------- .../provider/flashduty_provider_test.go | 12 +++++----- 2 files changed, 11 insertions(+), 23 deletions(-) diff --git a/alert/sender/provider/flashduty_provider.go b/alert/sender/provider/flashduty_provider.go index 34fc889d0..8b9108b71 100644 --- a/alert/sender/provider/flashduty_provider.go +++ b/alert/sender/provider/flashduty_provider.go @@ -38,10 +38,6 @@ func (p *FlashDutyProvider) Notify(ctx context.Context, req *NotifyRequest) *Not if req.Config.RequestConfig == nil || req.Config.RequestConfig.FlashDutyRequestConfig == nil { return &NotifyResult{Target: "", Response: "", Err: errors.New("flashduty request config not found")} } - events, err := prepareFlashDutyEvents(req.Config.RequestConfig.FlashDutyRequestConfig, req.Events, req.HttpClient) - if err != nil { - return &NotifyResult{Err: err} - } channelIDs := req.FlashDutyChannelIDs if len(channelIDs) == 0 { channelIDs = []int64{0} @@ -52,7 +48,7 @@ func (p *FlashDutyProvider) Notify(ctx context.Context, req *NotifyRequest) *Not for _, channelID := range channelIDs { target := strconv.FormatInt(channelID, 10) targets = append(targets, target) - resp, err := sendFlashDutyEvents(req.Config.RequestConfig.FlashDutyRequestConfig, events, channelID, req.HttpClient) + resp, err := SendFlashDuty(req.Config.RequestConfig.FlashDutyRequestConfig, req.Events, channelID, req.HttpClient) if err != nil { // 单个 channel 失败不要中断其他 channel 的发送 failedMsgs = append(failedMsgs, fmt.Sprintf("channel %s: %v", target, err)) @@ -68,14 +64,6 @@ func (p *FlashDutyProvider) Notify(ctx context.Context, req *NotifyRequest) *Not return &NotifyResult{Target: strings.Join(targets, ","), Response: strings.Join(responses, "; "), Err: aggErr} } -func SendFlashDuty(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, flashDutyChannelID int64, client *http.Client) (string, error) { - events, err := prepareFlashDutyEvents(config, events, client) - if err != nil { - return "", err - } - return sendFlashDutyEvents(config, events, flashDutyChannelID, client) -} - // Prepare private event copies so other notification channels can still use the screenshots. func prepareFlashDutyEvents(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, client *http.Client) ([]*models.AlertCurEvent, error) { if client == nil { @@ -202,10 +190,10 @@ func uploadFlashDutyImage(integrationURL, imageBase64 string, client *http.Clien return result.Data.ImageKey, nil } -func sendFlashDutyEvents(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, flashDutyChannelID int64, client *http.Client) (string, error) { - // todo 每一个 channel 批量发送事件 - if client == nil { - return "", fmt.Errorf("http client not found") +func SendFlashDuty(config *models.FlashDutyRequestConfig, events []*models.AlertCurEvent, flashDutyChannelID int64, client *http.Client) (string, error) { + events, err := prepareFlashDutyEvents(config, events, client) + if err != nil { + return "", err } body, err := json.Marshal(events) diff --git a/alert/sender/provider/flashduty_provider_test.go b/alert/sender/provider/flashduty_provider_test.go index e4e8b31f7..1d32f433c 100644 --- a/alert/sender/provider/flashduty_provider_test.go +++ b/alert/sender/provider/flashduty_provider_test.go @@ -1,12 +1,12 @@ package provider import ( - "os" "context" "encoding/base64" "encoding/json" "io" - + "os" + "time" "net/http" @@ -44,8 +44,8 @@ func TestFlashDutyUploadBeforeNotify(t *testing.T) { return } sends++ - if uploads != 1 { - t.Errorf("uploads before send: %d", uploads) + if uploads != sends { + t.Errorf("uploads before send: got %d, want %d", uploads, sends) } if r.URL.Path != "/event/push" || r.URL.Query().Get("channel_id") == "" { t.Errorf("unexpected event URL: %s", r.URL) @@ -83,7 +83,7 @@ func TestFlashDutyUploadBeforeNotify(t *testing.T) { if result.Err != nil { t.Fatal(result.Err) } - if uploads != 1 || sends != 2 { + if uploads != 2 || sends != 2 { t.Fatalf("uploads=%d sends=%d", uploads, sends) } if len(original.ShotImageBase64) != 1 || len(original.RuleConfigJson.(map[string]interface{})) != 1 { @@ -221,7 +221,7 @@ func TestFlashDutyOptionalImageFailures(t *testing.T) { if result.Err != nil || attempts != 3 || channels["1"] != 1 || channels["2"] != 1 { t.Fatalf("result=%+v attempts=%d channels=%v", result, attempts, channels) } - wantUploads := 1 + wantUploads := 2 if failure == "non-object config" || failure == "malformed stored config" { wantUploads = 0 } From 6cddd24b7916e19e45132e793c84c34f5bf4a4e9 Mon Sep 17 00:00:00 2001 From: liangyanbin Date: Mon, 26 Oct 2026 20:18:36 +0800 Subject: [PATCH 3/3] feat(alert): parallelize FlashDuty image uploads with bounded concurrency --- alert/sender/provider/flashduty_provider.go | 39 +++++++++------ .../provider/flashduty_provider_test.go | 49 +++++++++++++++++++ 2 files changed, 74 insertions(+), 14 deletions(-) diff --git a/alert/sender/provider/flashduty_provider.go b/alert/sender/provider/flashduty_provider.go index 8b9108b71..1908bab2e 100644 --- a/alert/sender/provider/flashduty_provider.go +++ b/alert/sender/provider/flashduty_provider.go @@ -13,6 +13,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" "github.com/ccfos/nightingale/v6/models" @@ -70,6 +71,9 @@ func prepareFlashDutyEvents(config *models.FlashDutyRequestConfig, events []*mod return nil, errors.New("http client not found") } prepared := make([]*models.AlertCurEvent, len(events)) + var uploads sync.WaitGroup + // Bound concurrent uploads to avoid excessive memory usage for large batches. + uploadSlots := make(chan struct{}, 5) for i, event := range events { if event == nil { continue @@ -107,23 +111,30 @@ func prepareFlashDutyEvents(config *models.FlashDutyRequestConfig, events []*mod if ruleConfig == nil { ruleConfig = make(map[string]json.RawMessage) } - imageKey, err := uploadFlashDutyImage(config.IntegrationUrl, imageBase64, client) - if err != nil { - // Screenshots are optional: keep sending alerts even if uploading fails. - logger.Warningf("send_flashduty: upload image failed event_id=%d: %v", event.Id, err) - break - } - // Match NewNorthstarAlertEvent: annotations.img_key and a string snapshot_pic_url. - ruleConfig["snapshot_pic_url"], _ = json.Marshal(imageKey) - copied.RuleConfigJson = ruleConfig - copied.AnnotationsJSON = make(map[string]string, len(event.AnnotationsJSON)+1) - for name, value := range event.AnnotationsJSON { - copied.AnnotationsJSON[name] = value - } - copied.AnnotationsJSON["img_key"] = imageKey + uploadSlots <- struct{}{} + uploads.Add(1) + go func() { + defer uploads.Done() + defer func() { <-uploadSlots }() + imageKey, err := uploadFlashDutyImage(config.IntegrationUrl, imageBase64, client) + if err != nil { + // Screenshots are optional: keep sending alerts even if uploading fails. + logger.Warningf("send_flashduty: upload image failed event_id=%d: %v", event.Id, err) + return + } + // Match NewNorthstarAlertEvent: annotations.img_key and a string snapshot_pic_url. + ruleConfig["snapshot_pic_url"], _ = json.Marshal(imageKey) + copied.RuleConfigJson = ruleConfig + copied.AnnotationsJSON = make(map[string]string, len(event.AnnotationsJSON)+1) + for name, value := range event.AnnotationsJSON { + copied.AnnotationsJSON[name] = value + } + copied.AnnotationsJSON["img_key"] = imageKey + }() break } } + uploads.Wait() return prepared, nil } diff --git a/alert/sender/provider/flashduty_provider_test.go b/alert/sender/provider/flashduty_provider_test.go index 1d32f433c..12b21494c 100644 --- a/alert/sender/provider/flashduty_provider_test.go +++ b/alert/sender/provider/flashduty_provider_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/base64" "encoding/json" + "fmt" "io" "os" @@ -13,6 +14,7 @@ import ( "net/http/httptest" "reflect" "strings" + "sync/atomic" "testing" "github.com/ccfos/nightingale/v6/models" @@ -91,6 +93,53 @@ func TestFlashDutyUploadBeforeNotify(t *testing.T) { } } +func TestFlashDutyConcurrentImageUploads(t *testing.T) { + var started atomic.Int32 + allStarted := make(chan struct{}) + client := &http.Client{Transport: flashDutyTestTransport(func(r *http.Request) (*http.Response, error) { + if started.Add(1) == 2 { + close(allStarted) + } + select { + case <-allStarted: + case <-time.After(5 * time.Second): + return nil, fmt.Errorf("image uploads did not run concurrently") + } + file, _, err := r.FormFile("image") + if err != nil { + return nil, err + } + defer file.Close() + data, err := io.ReadAll(file) + if err != nil { + return nil, err + } + body := fmt.Sprintf(`{"data":{"image_key":"img_%s"}}`, data) + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body))}, nil + })} + events := []*models.AlertCurEvent{ + {Id: 1, ShotImageBase64: map[string]string{"panel": base64.StdEncoding.EncodeToString([]byte("first"))}}, + nil, + {Id: 2}, + {Id: 3, ShotImageBase64: map[string]string{"panel": base64.StdEncoding.EncodeToString([]byte("last"))}}, + } + prepared, err := prepareFlashDutyEvents(&models.FlashDutyRequestConfig{IntegrationUrl: "http://flashduty.test/push"}, events, client) + if err != nil { + t.Fatal(err) + } + if started.Load() != 2 || len(prepared) != len(events) || prepared[1] != nil || prepared[2].Id != 2 { + t.Fatalf("unexpected prepared events: %+v; uploads=%d", prepared, started.Load()) + } + for i, wantKey := range map[int]string{0: "img_first", 3: "img_last"} { + if prepared[i].Id != events[i].Id || prepared[i].AnnotationsJSON["img_key"] != wantKey { + t.Errorf("event %d: unexpected prepared event: %+v", i, prepared[i]) + } + if prepared[i].ShotImageBase64 != nil || len(events[i].ShotImageBase64) != 1 || events[i].AnnotationsJSON != nil { + t.Errorf("event %d: original changed or outgoing screenshot not cleared", i) + } + } +} + func TestFlashDutyImagePreparation(t *testing.T) { for _, tc := range []struct { name, response, image string