237 lines
7.5 KiB
Go
237 lines
7.5 KiB
Go
package ingest
|
||
|
||
import (
|
||
"bytes"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.apinb.com/ops/logs/internal/config"
|
||
)
|
||
|
||
const (
|
||
alertResponseBodyLimit = 1 << 20
|
||
maxAlertTraceIDLength = 96
|
||
)
|
||
|
||
var errAlertForwardDisabled = errors.New("Alert 转发未启用或 base_url 为空")
|
||
|
||
type alertForwardResponse struct {
|
||
Code *int32 `json:"code"`
|
||
Details json.RawMessage `json:"details"`
|
||
}
|
||
|
||
// 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"`
|
||
State string `json:"state,omitempty"`
|
||
SourceEventKey string `json:"source_event_key"`
|
||
TraceID string `json:"trace_id"`
|
||
OccurredAt time.Time `json:"occurred_at"`
|
||
RawData json.RawMessage `json:"raw_data"`
|
||
}
|
||
|
||
type RawEventIngestBody struct {
|
||
SourceType string `json:"source_type"`
|
||
SourceEventKey string `json:"source_event_key"`
|
||
State string `json:"state,omitempty"`
|
||
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 errAlertForwardDisabled
|
||
}
|
||
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
|
||
}
|
||
body.TraceID = ensureAlertTraceID(body.TraceID, body.SourceEventKey)
|
||
raw, err := json.Marshal(body)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
result, err := postAlertPayload(cfg, "/Alert/v1/alerts/receive", raw, body.TraceID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var details struct {
|
||
ID uint `json:"id"`
|
||
RawEventID uint `json:"raw_event_id"`
|
||
AlertRecordID uint `json:"alert_record_id"`
|
||
IncidentID uint `json:"incident_id"`
|
||
Status string `json:"status"`
|
||
Fingerprint string `json:"fingerprint"`
|
||
}
|
||
if err := json.Unmarshal(result.Details, &details); err != nil {
|
||
return fmt.Errorf("Alert 响应 details 无效：%v；请稍后重试", err)
|
||
}
|
||
alertRecordID := details.AlertRecordID
|
||
if alertRecordID == 0 {
|
||
alertRecordID = details.ID
|
||
}
|
||
if details.RawEventID == 0 || alertRecordID == 0 || details.IncidentID == 0 ||
|
||
(details.ID != 0 && details.AlertRecordID != 0 && details.ID != details.AlertRecordID) ||
|
||
(details.Status != "firing" && details.Status != "resolved") || !validFingerprint(details.Fingerprint) {
|
||
return fmt.Errorf("Alert 响应缺少完整写入结果，无法确认转发成功；请稍后重试")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte, traceID string) (*alertForwardResponse, error) {
|
||
req, err := http.NewRequest(http.MethodPost, cfg.BaseURL+path, bytes.NewReader(payload))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("创建 Alert 转发请求失败：%w", err)
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("X-Trace-Id", traceID)
|
||
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 nil, fmt.Errorf("发送 Alert 转发请求失败：%w", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
responseBody, err := io.ReadAll(io.LimitReader(resp.Body, alertResponseBodyLimit+1))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("读取 Alert 响应失败：%w", err)
|
||
}
|
||
if len(responseBody) > alertResponseBodyLimit {
|
||
return nil, fmt.Errorf("Alert 响应体超过 %d 字节限制，请稍后重试", alertResponseBodyLimit)
|
||
}
|
||
if resp.StatusCode != http.StatusOK {
|
||
return nil, fmt.Errorf("Alert 返回 HTTP %d，请稍后重试", resp.StatusCode)
|
||
}
|
||
|
||
var result alertForwardResponse
|
||
if err := json.Unmarshal(responseBody, &result); err != nil {
|
||
return nil, fmt.Errorf("Alert 响应不是有效 JSON：%v；请稍后重试", err)
|
||
}
|
||
if result.Code == nil {
|
||
return nil, fmt.Errorf("Alert 响应缺少 code，无法确认转发成功；请稍后重试")
|
||
}
|
||
if *result.Code != 0 {
|
||
return nil, fmt.Errorf("Alert 拒绝转发，业务 code=%d；请稍后重试", *result.Code)
|
||
}
|
||
if len(result.Details) == 0 || string(result.Details) == "null" || string(result.Details) == `""` {
|
||
return nil, fmt.Errorf("Alert 响应缺少 details，无法确认转发成功；请稍后重试")
|
||
}
|
||
return &result, 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,
|
||
SourceEventKey: body.SourceEventKey,
|
||
State: body.State,
|
||
ResourceUID: rawEventResourceUID(body.Labels),
|
||
EventTime: body.OccurredAt,
|
||
Severity: body.SeverityCode,
|
||
Title: firstNonEmpty(body.AlertName, "日志事件"),
|
||
Message: firstNonEmpty(body.Summary, body.Description),
|
||
Labels: body.Labels,
|
||
Annotations: annotations,
|
||
ParseStatus: parseStatus,
|
||
RawPayload: body.RawData,
|
||
TraceID: ensureAlertTraceID(body.TraceID, body.SourceEventKey),
|
||
}
|
||
}
|
||
|
||
func ensureAlertTraceID(traceID, sourceEventKey string) string {
|
||
traceID = strings.TrimSpace(traceID)
|
||
if traceID != "" && len(traceID) <= maxAlertTraceIDLength {
|
||
return traceID
|
||
}
|
||
sum := sha256.Sum256([]byte(strings.TrimSpace(sourceEventKey)))
|
||
return hex.EncodeToString(sum[:16])
|
||
}
|
||
|
||
func validFingerprint(value string) bool {
|
||
if len(value) != 64 {
|
||
return false
|
||
}
|
||
decoded, err := hex.DecodeString(value)
|
||
return err == nil && len(decoded) == 32
|
||
}
|
||
|
||
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 ""
|
||
}
|