package ingest import ( "bytes" "encoding/json" "fmt" "io" "net/http" "strings" "time" "git.apinb.com/ops/logs/internal/config" ) const alertResponseBodyLimit = 1 << 20 // AlertReceiveBody 与 alert ReceiveRequest 对齐(含必填 raw_data) type AlertReceiveBody struct { AlertName string `json:"alert_name"` Summary string `json:"summary"` Description string `json:"description"` SeverityCode string `json:"severity_code"` Value string `json:"value"` Threshold string `json:"threshold"` Labels map[string]string `json:"labels"` Agent string `json:"agent"` PolicyID uint `json:"policy_id"` RawData json.RawMessage `json:"raw_data"` } type RawEventIngestBody struct { SourceType string `json:"source_type"` ResourceUID string `json:"resource_uid,omitempty"` EventTime time.Time `json:"event_time"` Severity string `json:"severity"` Title string `json:"title"` Message string `json:"message"` Labels map[string]string `json:"labels,omitempty"` Annotations map[string]string `json:"annotations,omitempty"` ParseStatus string `json:"parse_status"` RawPayload json.RawMessage `json:"raw_payload"` TraceID string `json:"trace_id,omitempty"` } func forwardAlert(body AlertReceiveBody) error { cfg := config.Spec.AlertForward if cfg == nil || !cfg.Enabled || cfg.BaseURL == "" { return nil } if len(body.RawData) == 0 { return fmt.Errorf("raw_data 不能为空") } if body.AlertName == "" { body.AlertName = "日志告警" } if body.PolicyID == 0 && cfg.DefaultPolicyID > 0 { body.PolicyID = cfg.DefaultPolicyID } raw, err := json.Marshal(body) if err != nil { return err } return postAlertPayload(cfg, "/Alert/v1/alerts/receive", 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 { return fmt.Errorf("创建 Alert 转发请求失败:%w", 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, CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }, } resp, err := client.Do(req) if err != nil { return fmt.Errorf("发送 Alert 转发请求失败:%w", err) } 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 { 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 } func buildRawEventIngestBody(body AlertReceiveBody, parseStatus string) RawEventIngestBody { sourceType := rawEventSourceType(body) if parseStatus == "" { parseStatus = "parsed" } annotations := map[string]string{ "description": body.Description, "value": body.Value, "threshold": body.Threshold, "agent": body.Agent, } return RawEventIngestBody{ SourceType: sourceType, ResourceUID: rawEventResourceUID(body.Labels), EventTime: time.Now().UTC(), Severity: body.SeverityCode, Title: firstNonEmpty(body.AlertName, "日志事件"), Message: firstNonEmpty(body.Summary, body.Description), Labels: body.Labels, Annotations: annotations, ParseStatus: parseStatus, RawPayload: body.RawData, } } func rawEventSourceType(body AlertReceiveBody) string { if body.Labels != nil { switch strings.TrimSpace(body.Labels["source_subtype"]) { case "syslog": return "syslog" case "snmp_trap": return "trap" } } switch body.Agent { case "logs-syslog": return "syslog" case "logs-trap": return "trap" default: return "syslog" } } func rawEventResourceUID(labels map[string]string) string { if labels == nil { return "" } if uid := strings.TrimSpace(labels["resource_uid"]); uid != "" { return uid } category := strings.TrimSpace(labels["resource_category"]) identity := strings.TrimSpace(labels["service_identity"]) if category != "" && identity != "" { return category + ":" + identity } return "" }