fix: honor alert forwarding business responses

This commit is contained in:
zxr
2026-07-18 08:51:58 +08:00
parent 601bfc58aa
commit d381360dfb
2 changed files with 33 additions and 26 deletions

View File

@@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io"
"net/http" "net/http"
"strings" "strings"
"time" "time"
@@ -11,6 +12,8 @@ import (
"git.apinb.com/ops/logs/internal/config" "git.apinb.com/ops/logs/internal/config"
) )
const alertResponseBodyLimit = 1 << 20
// AlertReceiveBody 与 alert ReceiveRequest 对齐(含必填 raw_data // AlertReceiveBody 与 alert ReceiveRequest 对齐(含必填 raw_data
type AlertReceiveBody struct { type AlertReceiveBody struct {
AlertName string `json:"alert_name"` AlertName string `json:"alert_name"`
@@ -57,10 +60,13 @@ func forwardAlert(body AlertReceiveBody) error {
if err != nil { if err != nil {
return err return err
} }
url := cfg.BaseURL + "/Alert/v1/alerts/receive" return postAlertPayload(cfg, "/Alert/v1/alerts/receive", raw)
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(raw)) }
func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte) error {
req, err := http.NewRequest(http.MethodPost, cfg.BaseURL+path, bytes.NewReader(payload))
if err != nil { if err != nil {
return err return fmt.Errorf("创建 Alert 转发请求失败:%w", err)
} }
req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Type", "application/json")
if cfg.InternalKey != "" { if cfg.InternalKey != "" {
@@ -69,11 +75,32 @@ func forwardAlert(body AlertReceiveBody) error {
client := &http.Client{Timeout: 10 * time.Second} client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req) resp, err := client.Do(req)
if err != nil { if err != nil {
return err return fmt.Errorf("发送 Alert 转发请求失败:%w", err)
} }
defer resp.Body.Close() defer resp.Body.Close()
responseBody, err := io.ReadAll(io.LimitReader(resp.Body, alertResponseBodyLimit+1))
if err != nil {
return fmt.Errorf("读取 Alert 响应失败:%w", err)
}
if len(responseBody) > alertResponseBodyLimit {
return fmt.Errorf("Alert 响应体超过 %d 字节限制,请稍后重试", alertResponseBodyLimit)
}
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
return fmt.Errorf("alert returned HTTP %d", resp.StatusCode) return fmt.Errorf("Alert 返回 HTTP %d,请稍后重试", resp.StatusCode)
}
var result struct {
Code *int32 `json:"code"`
}
if err := json.Unmarshal(responseBody, &result); err != nil {
return fmt.Errorf("Alert 响应不是有效 JSON%v请稍后重试", err)
}
if result.Code == nil {
return fmt.Errorf("Alert 响应缺少 code无法确认转发成功请稍后重试")
}
if *result.Code != 0 {
return fmt.Errorf("Alert 拒绝转发,业务 code=%d请稍后重试", *result.Code)
} }
return nil return nil
} }

View File

@@ -1,10 +1,8 @@
package ingest package ingest
import ( import (
"bytes"
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http"
"strings" "strings"
"time" "time"
@@ -118,25 +116,7 @@ func forwardRawEvent(body RawEventIngestBody) error {
if err != nil { if err != nil {
return err return err
} }
url := cfg.BaseURL + "/Alert/v1/raw-events/ingest" return postAlertPayload(cfg, "/Alert/v1/raw-events/ingest", raw)
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(raw))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
if cfg.InternalKey != "" {
req.Header.Set("X-Internal-Key", cfg.InternalKey)
}
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("alert returned HTTP %d", resp.StatusCode)
}
return nil
} }
func markOutboxRetry(row models.AlertOutbox, msg string) { func markOutboxRetry(row models.AlertOutbox, msg string) {