From 6e66bad884c4731b3a70c5c917eb052de89db975 Mon Sep 17 00:00:00 2001 From: zxr <271055687@qq.com> Date: Tue, 21 Jul 2026 02:32:18 +0800 Subject: [PATCH] fix: make log alert delivery durable --- internal/ingest/alert_forward.go | 141 ++++++++----- internal/ingest/alert_outbox.go | 270 ++++++++++++++++++------- internal/ingest/engine.go | 305 ++++++++++++++++++++--------- internal/ingest/replay.go | 91 +++++++-- internal/logic/controllers/crud.go | 4 - internal/models/alert_outbox.go | 60 +++--- internal/models/log_event.go | 94 ++++----- internal/models/query.go | 49 +++++ 8 files changed, 709 insertions(+), 305 deletions(-) diff --git a/internal/ingest/alert_forward.go b/internal/ingest/alert_forward.go index 6ffe14d..c628df4 100644 --- a/internal/ingest/alert_forward.go +++ b/internal/ingest/alert_forward.go @@ -2,7 +2,9 @@ package ingest import ( "bytes" + "encoding/hex" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -14,38 +16,50 @@ import ( const alertResponseBodyLimit = 1 << 20 +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"` - RawData json.RawMessage `json:"raw_data"` + 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"` + OccurredAt time.Time `json:"occurred_at"` + 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"` + 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 nil + return errAlertForwardDisabled } if len(body.RawData) == 0 { return fmt.Errorf("raw_data 不能为空") @@ -60,13 +74,37 @@ func forwardAlert(body AlertReceiveBody) error { if err != nil { return err } - return postAlertPayload(cfg, "/Alert/v1/alerts/receive", raw) + result, err := postAlertPayload(cfg, "/Alert/v1/alerts/receive", raw) + 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) error { +func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte) (*alertForwardResponse, error) { req, err := http.NewRequest(http.MethodPost, cfg.BaseURL+path, bytes.NewReader(payload)) if err != nil { - return fmt.Errorf("创建 Alert 转发请求失败：%w", err) + return nil, fmt.Errorf("创建 Alert 转发请求失败：%w", err) } req.Header.Set("Content-Type", "application/json") if cfg.InternalKey != "" { @@ -80,34 +118,35 @@ func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte) } resp, err := client.Do(req) if err != nil { - return fmt.Errorf("发送 Alert 转发请求失败：%w", err) + 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 fmt.Errorf("读取 Alert 响应失败：%w", err) + return nil, fmt.Errorf("读取 Alert 响应失败：%w", err) } if len(responseBody) > alertResponseBodyLimit { - return fmt.Errorf("Alert 响应体超过 %d 字节限制，请稍后重试", alertResponseBodyLimit) + return nil, fmt.Errorf("Alert 响应体超过 %d 字节限制，请稍后重试", alertResponseBodyLimit) } if resp.StatusCode != http.StatusOK { - return fmt.Errorf("Alert 返回 HTTP %d，请稍后重试", resp.StatusCode) + return nil, fmt.Errorf("Alert 返回 HTTP %d，请稍后重试", resp.StatusCode) } - var result struct { - Code *int32 `json:"code"` - } + var result alertForwardResponse if err := json.Unmarshal(responseBody, &result); err != nil { - return fmt.Errorf("Alert 响应不是有效 JSON：%v；请稍后重试", err) + return nil, fmt.Errorf("Alert 响应不是有效 JSON：%v；请稍后重试", err) } if result.Code == nil { - return fmt.Errorf("Alert 响应缺少 code，无法确认转发成功；请稍后重试") + return nil, fmt.Errorf("Alert 响应缺少 code，无法确认转发成功；请稍后重试") } if *result.Code != 0 { - return fmt.Errorf("Alert 拒绝转发，业务 code=%d；请稍后重试", *result.Code) + return nil, fmt.Errorf("Alert 拒绝转发，业务 code=%d；请稍后重试", *result.Code) } - return nil + 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 { @@ -122,19 +161,29 @@ func buildRawEventIngestBody(body AlertReceiveBody, parseStatus string) RawEvent "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, + 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, } } +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"]) { diff --git a/internal/ingest/alert_outbox.go b/internal/ingest/alert_outbox.go index b349c60..21f64d5 100644 --- a/internal/ingest/alert_outbox.go +++ b/internal/ingest/alert_outbox.go @@ -2,98 +2,179 @@ package ingest import ( "encoding/json" + "errors" "fmt" + "log" + "os" "strings" "time" "git.apinb.com/ops/logs/internal/config" "git.apinb.com/ops/logs/internal/impl" "git.apinb.com/ops/logs/internal/models" + "gorm.io/gorm" + "gorm.io/gorm/clause" ) const ( - outboxStatusPending = "pending" - outboxStatusRetrying = "retrying" - outboxStatusSent = "sent" - outboxStatusDead = "dead" + outboxStatusPending = "pending" + outboxStatusProcessing = "processing" + outboxStatusRetrying = "retrying" + outboxStatusSent = "sent" + outboxStatusDead = "dead" ) -func enqueueAlert(logEventID uint, body AlertReceiveBody) error { +func enqueueAlertWithDB(db *gorm.DB, logEventID uint, body AlertReceiveBody) (uint, error) { payload, err := json.Marshal(body) if err != nil { - return err + return 0, err } - return enqueuePayload(logEventID, string(payload)) + return enqueuePayloadWithDB(db, logEventID, string(payload)) } -func enqueueRawEvent(logEventID uint, body AlertReceiveBody, parseStatus string) error { +func enqueueRawEventWithDB(db *gorm.DB, logEventID uint, body AlertReceiveBody, parseStatus string) (uint, error) { payload, err := json.Marshal(buildRawEventIngestBody(body, parseStatus)) if err != nil { - return err + return 0, err } - return enqueuePayload(logEventID, string(payload)) + return enqueuePayloadWithDB(db, logEventID, string(payload)) } -func enqueuePayload(logEventID uint, payloadJSON string) error { +func enqueuePayloadWithDB(db *gorm.DB, logEventID uint, payloadJSON string) (uint, error) { + if db == nil { + return 0, fmt.Errorf("database is not initialized") + } + now, err := databaseNow(db) + if err != nil { + return 0, err + } row := models.AlertOutbox{ LogEventID: logEventID, PayloadJSON: payloadJSON, Status: outboxStatusPending, RetryCount: 0, - NextRetryAt: time.Now(), + NextRetryAt: now, LastError: "", } - return impl.DBService.Create(&row).Error + if err := db.Create(&row).Error; err != nil { + return 0, err + } + return row.ID, nil } func StartAlertDispatcher() { + owner := dispatcherOwner() go func() { ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() for range ticker.C { - processAlertOutboxBatch(20) + if _, err := ProcessAlertOutboxBatch(impl.DBService, 20, owner); err != nil { + log.Printf("logs: alert outbox dispatch: %v", err) + } } }() } -func processAlertOutboxBatch(limit int) { +func dispatcherOwner() string { + host, _ := os.Hostname() + return fmt.Sprintf("%s:%d:%d", host, os.Getpid(), time.Now().UnixNano()) +} + +// ProcessAlertOutboxBatch 原子领取并处理一批任务，可安全用于多实例 worker。 +func ProcessAlertOutboxBatch(db *gorm.DB, limit int, owner string) (int, error) { + if db == nil { + return 0, fmt.Errorf("database is not initialized") + } if limit <= 0 { limit = 20 } - var rows []models.AlertOutbox - now := time.Now() - err := impl.DBService. - Where("status IN ? AND next_retry_at <= ?", []string{outboxStatusPending, outboxStatusRetrying}, now). - Order("id asc"). - Limit(limit). - Find(&rows).Error - if err != nil || len(rows) == 0 { - return + owner = strings.TrimSpace(owner) + if owner == "" { + return 0, fmt.Errorf("outbox lease owner is required") } - for _, row := range rows { - processOneOutbox(row) + processed := 0 + for processed < limit { + rows, err := claimAlertOutboxBatch(db, 1, owner) + if err != nil { + return processed, err + } + if len(rows) == 0 { + break + } + if err := processOneOutbox(db, rows[0]); err != nil { + return processed, err + } + processed++ } + return processed, nil } -func processOneOutbox(row models.AlertOutbox) { +func claimAlertOutboxBatch(db *gorm.DB, limit int, owner string) ([]models.AlertOutbox, error) { + var rows []models.AlertOutbox + now, err := databaseNow(db) + if err != nil { + return nil, err + } + leaseUntil := now.Add(30 * time.Second) + err = db.Transaction(func(tx *gorm.DB) error { + if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}). + Where("(status IN ? AND next_retry_at <= ?) OR (status = ? AND lease_until IS NOT NULL AND lease_until <= ?)", []string{outboxStatusPending, outboxStatusRetrying}, now, outboxStatusProcessing, now). + Order("id asc").Limit(limit).Find(&rows).Error; err != nil { + return err + } + if len(rows) == 0 { + return nil + } + ids := make([]uint, 0, len(rows)) + for i := range rows { + ids = append(ids, rows[i].ID) + } + if err := tx.Model(&models.AlertOutbox{}).Where("id IN ?", ids).Updates(map[string]interface{}{ + "status": outboxStatusProcessing, + "lease_until": leaseUntil, + "lease_owner": owner, + }).Error; err != nil { + return err + } + for i := range rows { + rows[i].Status = outboxStatusProcessing + rows[i].LeaseUntil = &leaseUntil + rows[i].LeaseOwner = owner + } + return nil + }) + return rows, err +} + +func processOneOutbox(db *gorm.DB, row models.AlertOutbox) error { var body AlertReceiveBody if err := json.Unmarshal([]byte(row.PayloadJSON), &body); err != nil { - markOutboxDead(row.ID, row.RetryCount, "invalid_payload: "+err.Error()) - return + now, nowErr := databaseNow(db) + if nowErr != nil { + return nowErr + } + return markOutboxDead(db, row, row.RetryCount, "invalid_payload: "+err.Error(), now) } - if err := forwardOutboxPayload(row.PayloadJSON, body); err != nil { - markOutboxRetry(row, err.Error()) - return + forwardErr := forwardOutboxPayload(row.PayloadJSON, body) + now, err := databaseNow(db) + if err != nil { + return err } - _ = impl.DBService.Model(&models.AlertOutbox{}).Where("id = ?", row.ID).Updates(map[string]interface{}{ - "status": outboxStatusSent, - "last_error": "", - "next_retry_at": time.Now(), - }).Error - _ = impl.DBService.Model(&models.LogEvent{}).Where("id = ?", row.LogEventID).Updates(map[string]interface{}{ - "alert_sent": true, - "dispatch_status": "sent", - }).Error + if forwardErr != nil { + if errors.Is(forwardErr, errAlertForwardDisabled) { + return markOutboxDead(db, row, row.RetryCount, forwardErr.Error(), now) + } + return markOutboxRetry(db, row, forwardErr.Error(), now) + } + return markOutboxSent(db, row, now) +} + +func databaseNow(db *gorm.DB) (time.Time, error) { + var now time.Time + if err := db.Raw("SELECT CURRENT_TIMESTAMP").Scan(&now).Error; err != nil { + return time.Time{}, err + } + return now.UTC(), nil } func forwardOutboxPayload(payloadJSON string, legacyBody AlertReceiveBody) error { @@ -107,7 +188,7 @@ func forwardOutboxPayload(payloadJSON string, legacyBody AlertReceiveBody) error func forwardRawEvent(body RawEventIngestBody) error { cfg := config.Spec.AlertForward if cfg == nil || !cfg.Enabled || cfg.BaseURL == "" { - return nil + return errAlertForwardDisabled } if len(body.RawPayload) == 0 { return fmt.Errorf("raw_payload 不能为空") @@ -116,46 +197,103 @@ func forwardRawEvent(body RawEventIngestBody) error { if err != nil { return err } - return postAlertPayload(cfg, "/Alert/v1/raw-events/ingest", raw) + result, err := postAlertPayload(cfg, "/Alert/v1/raw-events/ingest", raw) + 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"` + ParseStatus string `json:"parse_status"` + Status string `json:"status"` + Fingerprint string `json:"fingerprint"` + } + if err := json.Unmarshal(result.Details, &details); err != nil { + return fmt.Errorf("Alert 原始事件响应 details 无效：%v；请稍后重试", err) + } + if body.ParseStatus == "unparsed" { + if details.ID == 0 || (details.ParseStatus != "unparsed" && details.ParseStatus != "replayed") { + return fmt.Errorf("Alert 原始事件响应缺少持久化结果，无法确认转发成功；请稍后重试") + } + return nil + } + rawEventID := details.RawEventID + if rawEventID == 0 { + rawEventID = details.ID + } + if rawEventID == 0 || details.AlertRecordID == 0 || details.IncidentID == 0 || + (details.ID != 0 && details.RawEventID != 0 && details.ID != details.RawEventID) || + (details.Status != "firing" && details.Status != "resolved") || !validFingerprint(details.Fingerprint) { + return fmt.Errorf("Alert 原始事件响应缺少完整处理结果，无法确认转发成功；请稍后重试") + } + return nil } -func markOutboxRetry(row models.AlertOutbox, msg string) { +func markOutboxSent(db *gorm.DB, row models.AlertOutbox, now time.Time) error { + return db.Transaction(func(tx *gorm.DB) error { + result := tx.Model(&models.AlertOutbox{}). + Where("id = ? AND status = ? AND lease_owner = ?", row.ID, outboxStatusProcessing, row.LeaseOwner). + Updates(map[string]interface{}{"status": outboxStatusSent, "last_error": "", "next_retry_at": now, "lease_until": nil, "lease_owner": ""}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return nil + } + return tx.Model(&models.LogEvent{}).Where("id = ? AND dispatch_outbox_id = ?", row.LogEventID, row.ID).Updates(map[string]interface{}{ + "alert_sent": true, "dispatch_status": "sent", + }).Error + }) +} + +func markOutboxRetry(db *gorm.DB, row models.AlertOutbox, msg string, now time.Time) error { retry := row.RetryCount + 1 const maxRetry = 5 if retry > maxRetry { - markOutboxDead(row.ID, retry, msg) - return + return markOutboxDead(db, row, retry, msg, now) } backoff := time.Duration(retry*retry) * time.Second if backoff > 60*time.Second { backoff = 60 * time.Second } - _ = impl.DBService.Model(&models.AlertOutbox{}).Where("id = ?", row.ID).Updates(map[string]interface{}{ - "status": outboxStatusRetrying, - "retry_count": retry, - "next_retry_at": time.Now().Add(backoff), - "last_error": truncateError(msg, 1024), - }).Error - _ = impl.DBService.Model(&models.LogEvent{}).Where("id = ?", row.LogEventID).Update("dispatch_status", "retrying").Error + return updateClaimedOutbox(db, row, map[string]interface{}{ + "status": outboxStatusRetrying, "retry_count": retry, "next_retry_at": now.Add(backoff), + "last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "", + }, "retrying") } -func markOutboxDead(id uint, retry int, msg string) { - _ = impl.DBService.Model(&models.AlertOutbox{}).Where("id = ?", id).Updates(map[string]interface{}{ - "status": outboxStatusDead, - "retry_count": retry, - "next_retry_at": time.Now(), - "last_error": truncateError(msg, 1024), - }).Error - var row models.AlertOutbox - if err := impl.DBService.Select("log_event_id").First(&row, id).Error; err == nil && row.LogEventID > 0 { - _ = impl.DBService.Model(&models.LogEvent{}).Where("id = ?", row.LogEventID).Update("dispatch_status", "dead").Error - } +func markOutboxDead(db *gorm.DB, row models.AlertOutbox, retry int, msg string, now time.Time) error { + return updateClaimedOutbox(db, row, map[string]interface{}{ + "status": outboxStatusDead, "retry_count": retry, "next_retry_at": now, + "last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "", + }, "dead") +} + +func updateClaimedOutbox(db *gorm.DB, row models.AlertOutbox, updates map[string]interface{}, eventStatus string) error { + return db.Transaction(func(tx *gorm.DB) error { + result := tx.Model(&models.AlertOutbox{}). + Where("id = ? AND status = ? AND lease_owner = ?", row.ID, outboxStatusProcessing, row.LeaseOwner). + Updates(updates) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return nil + } + return tx.Model(&models.LogEvent{}).Where("id = ? AND dispatch_outbox_id = ?", row.LogEventID, row.ID).Update("dispatch_status", eventStatus).Error + }) } func truncateError(s string, n int) string { s = strings.TrimSpace(s) - if len(s) <= n { + if n <= 0 { + return "" + } + runes := []rune(s) + if len(runes) <= n { return s } - return s[:n] + return string(runes[:n]) } diff --git a/internal/ingest/engine.go b/internal/ingest/engine.go index 4e23ad4..52a550b 100644 --- a/internal/ingest/engine.go +++ b/internal/ingest/engine.go @@ -3,6 +3,7 @@ package ingest import ( "encoding/json" "fmt" + "log" "net" "regexp" "sort" @@ -15,6 +16,7 @@ import ( "git.apinb.com/ops/logs/internal/impl" "git.apinb.com/ops/logs/internal/models" "github.com/gosnmp/gosnmp" + "gorm.io/gorm" ) type Engine struct { @@ -29,9 +31,12 @@ type Engine struct { } type resourceRef struct { - ResourceType string - ResourceID string - ResourceName string + ResourceType string + ResourceUID string + ResourceCategory string + ServiceIdentity string + ResourceID string + ResourceName string } func resourceTypePriority(resourceType string) int { @@ -88,6 +93,12 @@ func (e *Engine) Refresh() error { ResourceID: m.ResourceID, ResourceName: m.ResourceName, } + var mappingLabels map[string]string + if err := json.Unmarshal([]byte(m.LabelsJSON), &mappingLabels); err == nil { + ref.ResourceUID = strings.TrimSpace(mappingLabels["resource_uid"]) + ref.ResourceCategory = strings.TrimSpace(mappingLabels["resource_category"]) + ref.ServiceIdentity = strings.TrimSpace(mappingLabels["service_identity"]) + } var ips []string if err := json.Unmarshal([]byte(m.IPsJSON), &ips); err == nil { for _, ip := range ips { @@ -146,6 +157,7 @@ func normOID(s string) string { } func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) { + occurredAt := time.Now().UTC() parsed := parseSyslogPayload(payload) device := parsed.Hostname if device == "" { @@ -193,35 +205,38 @@ func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) { } } - if err := impl.DBService.Create(&ev).Error; err != nil { - return - } - if matched == nil { rawBytes, mErr := json.Marshal(string(payload)) if mErr != nil { return } + labels := map[string]string{ + "source_type": "log", + "source_subtype": "syslog", + "device": device, + "remote_addr": addr.String(), + "ip": addr.IP.String(), + "instance": firstNonEmpty(device, addr.String()), + "job": "logs-syslog", + } + applyResourceIdentity(&ev, labels, "", ref, "syslog", addr.IP.String()) body := AlertReceiveBody{ AlertName: "未解析 Syslog", Summary: summary, Description: parsed.RawLine, SeverityCode: sev, Value: parsed.Message, - Labels: map[string]string{ - "source_type": "log", - "source_subtype": "syslog", - "device": device, - "remote_addr": addr.String(), - "ip": addr.IP.String(), - "instance": firstNonEmpty(device, addr.String()), - "job": "logs-syslog", - }, - Agent: "logs-syslog", - RawData: rawBytes, + Labels: labels, + Agent: "logs-syslog", + RawData: rawBytes, } - if err := enqueueRawEvent(ev.ID, body, "unparsed"); err == nil { - _ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error + if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) { + body.SourceEventKey = logSourceEventKey(stored, "raw") + body.OccurredAt = occurredAt + body.State = "firing" + return enqueueRawEventWithDB(tx, stored.ID, body, "unparsed") + }); err != nil { + log.Printf("logs: persist syslog raw event: %v", err) } return } @@ -241,33 +256,26 @@ func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) { if matchDetails.ResourceUID != "" { labels["resource_uid"] = matchDetails.ResourceUID } - rawObj := map[string]interface{}{ - "source": "syslog", - "received_at": time.Now().UTC().Format(time.RFC3339), - "source_ip": addr.IP.String(), - "rule_id": matched.ID, - "log_entry_id": ev.ID, - "raw_packet": string(payload), - "parsed": detailObj, - "match": matchDetails.Captures, - } - rawBytes, mErr := json.Marshal(rawObj) - if mErr != nil { - return - } - body := AlertReceiveBody{ - AlertName: matched.AlertName, - Summary: summary, - Description: summary, - SeverityCode: firstNonEmpty(matchDetails.SeverityCode, firstNonEmpty(matched.SeverityCode, sev)), - Value: parsed.Message, - Labels: labels, - Agent: "logs-syslog", - PolicyID: matched.PolicyID, - RawData: rawBytes, - } - if err := enqueueAlert(ev.ID, body); err == nil { - _ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error + applyResourceIdentity(&ev, labels, matchDetails.ResourceUID, ref, "syslog", addr.IP.String()) + if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) { + rawObj := map[string]interface{}{ + "source": "syslog", "received_at": occurredAt.Format(time.RFC3339), "source_ip": addr.IP.String(), + "rule_id": matched.ID, "log_entry_id": stored.ID, "raw_packet": string(payload), + "parsed": detailObj, "match": matchDetails.Captures, + } + rawBytes, err := json.Marshal(rawObj) + if err != nil { + return 0, err + } + body := AlertReceiveBody{ + AlertName: matched.AlertName, Summary: summary, Description: summary, + SeverityCode: firstNonEmpty(matchDetails.SeverityCode, firstNonEmpty(matched.SeverityCode, sev)), + Value: parsed.Message, Labels: labels, Agent: "logs-syslog", PolicyID: matched.PolicyID, + State: "firing", SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes, + } + return enqueueAlertWithDB(tx, stored.ID, body) + }); err != nil { + log.Printf("logs: persist matched syslog event: %v", err) } } @@ -372,9 +380,12 @@ func extractWithNamedRegex(pattern, groupName, message, rawLine string) string { func normalizeExtractedResourceUID(uid string) string { uid = strings.TrimSpace(uid) - if uid == "" || strings.Contains(uid, ":") { + if uid == "" { return uid } + if category, identity, ok := splitKnownResourceUID(uid); ok { + return category + ":" + identity + } return "network:" + uid } @@ -445,6 +456,7 @@ func lookupTrapDict(e *Engine, trapOID string) *models.TrapDictionaryEntry { } func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) { + occurredAt := time.Now().UTC() trapOID := extractTrapOID(pkt) if trapShielded(e, addr, trapOID, pkt) { return @@ -490,10 +502,6 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) { SeverityCode: sev, TrapOID: trapOID, } - if err := impl.DBService.Create(&ev).Error; err != nil { - return - } - e.mu.RLock() rules := e.trapRules e.mu.RUnlock() @@ -504,26 +512,28 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) { if mErr != nil { return } + labels := map[string]string{ + "source_type": "log", "source_subtype": "snmp_trap", "trap_oid": trapOID, + "remote_addr": addr.String(), "ip": addr.IP.String(), "instance": addr.IP.String(), "job": "logs-trap", + } + applyResourceIdentity(&ev, labels, "", ref, "snmp_trap", addr.IP.String()) body := AlertReceiveBody{ AlertName: "未解析 SNMP Trap", Summary: readable, Description: fp, SeverityCode: sev, Value: string(vbJSON), - Labels: map[string]string{ - "source_type": "log", - "source_subtype": "snmp_trap", - "trap_oid": trapOID, - "remote_addr": addr.String(), - "ip": addr.IP.String(), - "instance": addr.IP.String(), - "job": "logs-trap", - }, - Agent: "logs-trap", - RawData: rawBytes, + Labels: labels, + Agent: "logs-trap", + RawData: rawBytes, } - if err := enqueueRawEvent(ev.ID, body, "unparsed"); err == nil { - _ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error + if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) { + body.State = "firing" + body.SourceEventKey = logSourceEventKey(stored, "raw") + body.OccurredAt = occurredAt + return enqueueRawEventWithDB(tx, stored.ID, body, "unparsed") + }); err != nil { + log.Printf("logs: persist trap raw event: %v", err) } return } @@ -551,6 +561,7 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) { labels["resource_id"] = trapOID } } + applyResourceIdentity(&ev, labels, "", ref, "snmp_trap", addr.IP.String()) resolved := map[string]interface{}{} if dict != nil { resolved["vendor"] = dict.Vendor @@ -561,36 +572,27 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) { resolved["severity_mapping"] = dict.SeverityMappingJSON resolved["parse_expression"] = dict.ParseExpression } - rawObj := map[string]interface{}{ - "source": "snmp_trap", - "received_at": time.Now().UTC().Format(time.RFC3339), - "source_ip": addr.IP.String(), - "log_entry_id": ev.ID, - "trap_oid": trapOID, - "varbinds": trapVarbinds(pkt), - "resolved": resolved, - "pdu_summary": fp, - } - if matched.ID != 0 { - rawObj["rule_id"] = matched.ID - } - rawBytes, mErr := json.Marshal(rawObj) - if mErr != nil { - return - } - body := AlertReceiveBody{ - AlertName: firstNonEmpty(matched.AlertName, "SNMP Trap"), - Summary: readable, - Description: desc, - SeverityCode: firstNonEmpty(matched.SeverityCode, sev), - Value: string(vbJSON), - Labels: labels, - Agent: "logs-trap", - PolicyID: matched.PolicyID, - RawData: rawBytes, - } - if err := enqueueAlert(ev.ID, body); err == nil { - _ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error + if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) { + rawObj := map[string]interface{}{ + "source": "snmp_trap", "received_at": occurredAt.Format(time.RFC3339), "source_ip": addr.IP.String(), + "log_entry_id": stored.ID, "trap_oid": trapOID, "varbinds": trapVarbinds(pkt), "resolved": resolved, "pdu_summary": fp, + } + if matched.ID != 0 { + rawObj["rule_id"] = matched.ID + } + rawBytes, err := json.Marshal(rawObj) + if err != nil { + return 0, err + } + body := AlertReceiveBody{ + AlertName: firstNonEmpty(matched.AlertName, "SNMP Trap"), Summary: readable, Description: desc, + SeverityCode: firstNonEmpty(matched.SeverityCode, sev), Value: string(vbJSON), Labels: labels, + Agent: "logs-trap", PolicyID: matched.PolicyID, State: "firing", + SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes, + } + return enqueueAlertWithDB(tx, stored.ID, body) + }); err != nil { + log.Printf("logs: persist matched trap event: %v", err) } } @@ -693,3 +695,116 @@ func (e *Engine) resolveResource(sourceIP, hostname string) (resourceRef, string } return resourceRef{}, "none" } + +func persistLogEventAndOutbox(ev *models.LogEvent, enqueue func(*gorm.DB, *models.LogEvent) (uint, error)) error { + if impl.DBService == nil { + return fmt.Errorf("database is not initialized") + } + if ev == nil || enqueue == nil { + return fmt.Errorf("log event and outbox builder are required") + } + return impl.DBService.Transaction(func(tx *gorm.DB) error { + if err := tx.Create(ev).Error; err != nil { + return err + } + outboxID, err := enqueue(tx, ev) + if err != nil { + return err + } + ev.DispatchStatus = "pending" + ev.DispatchOutboxID = outboxID + return tx.Model(ev).Updates(map[string]interface{}{ + "dispatch_status": ev.DispatchStatus, + "dispatch_outbox_id": ev.DispatchOutboxID, + }).Error + }) +} + +func logSourceEventKey(ev *models.LogEvent, purpose string) string { + return fmt.Sprintf("logs:%s:%d:%s", strings.TrimSpace(ev.SourceKind), ev.ID, strings.TrimSpace(purpose)) +} + +func applyResourceIdentity(ev *models.LogEvent, labels map[string]string, explicitUID string, ref resourceRef, subtype, sourceIP string) { + uid := strings.TrimSpace(explicitUID) + category := "" + identity := "" + if uid == "" { + uid = strings.TrimSpace(ref.ResourceUID) + } + if uid != "" { + if knownCategory, knownIdentity, ok := splitKnownResourceUID(uid); ok { + category, identity = knownCategory, knownIdentity + uid = category + ":" + identity + } else { + category = strings.ToLower(strings.TrimSpace(ref.ResourceCategory)) + if category == "" { + category = canonicalLogResourceCategory(ref.ResourceType, subtype) + } + identity = uid + uid = category + ":" + identity + } + } + if uid == "" && strings.TrimSpace(ref.ResourceCategory) != "" && strings.TrimSpace(ref.ServiceIdentity) != "" { + category = strings.ToLower(strings.TrimSpace(ref.ResourceCategory)) + identity = strings.TrimSpace(ref.ServiceIdentity) + uid = category + ":" + identity + } + if uid == "" && strings.TrimSpace(ref.ResourceID) != "" { + identity = strings.TrimSpace(ref.ResourceID) + category = canonicalLogResourceCategory(ref.ResourceType, subtype) + if knownCategory, knownIdentity, ok := splitKnownResourceUID(identity); ok { + category, identity = knownCategory, knownIdentity + uid = category + ":" + identity + } else { + uid = category + ":" + identity + } + } + if uid == "" { + category = canonicalLogResourceCategory("", subtype) + identity = strings.TrimSpace(sourceIP) + uid = category + ":" + identity + } + labels["resource_uid"] = uid + labels["resource_category"] = category + labels["service_identity"] = identity + ev.ResourceUID = uid + if ev.ResourceType == "" || ev.ResourceID == "" { + ev.ResourceType = category + ev.ResourceID = identity + } +} + +func canonicalLogResourceCategory(resourceType, subtype string) string { + switch strings.ToLower(strings.TrimSpace(resourceType)) { + case "server", "host": + return "host" + case "device", "network", "network_device": + return "network" + case "collector": + return "collector" + case "database", "middleware", "security", "storage", "room_device": + return strings.ToLower(strings.TrimSpace(resourceType)) + } + if strings.TrimSpace(subtype) == "snmp_trap" || strings.TrimSpace(subtype) == "trap" { + return "network" + } + return "log_source" +} + +func splitKnownResourceUID(value string) (string, string, bool) { + prefix, identity, found := strings.Cut(strings.TrimSpace(value), ":") + if !found { + return "", "", false + } + category := strings.ToLower(strings.TrimSpace(prefix)) + identity = strings.TrimSpace(identity) + if identity == "" { + return "", "", false + } + switch category { + case "host", "network", "collector", "database", "middleware", "security", "storage", "room_device", "log_source": + return category, identity, true + default: + return "", "", false + } +} diff --git a/internal/ingest/replay.go b/internal/ingest/replay.go index 37d7078..60d94b7 100644 --- a/internal/ingest/replay.go +++ b/internal/ingest/replay.go @@ -9,6 +9,7 @@ import ( "git.apinb.com/ops/logs/internal/impl" "git.apinb.com/ops/logs/internal/models" + "gorm.io/gorm" ) func BuildReplayRawEventPayload(ev models.LogEvent) (RawEventIngestBody, error) { @@ -16,9 +17,10 @@ func BuildReplayRawEventPayload(ev models.LogEvent) (RawEventIngestBody, error) if sourceType == "" { return RawEventIngestBody{}, fmt.Errorf("unsupported source kind %q", ev.SourceKind) } + now := time.Now().UTC() rawObj := map[string]interface{}{ "source": sourceType, - "replayed_at": time.Now().UTC().Format(time.RFC3339), + "replayed_at": now.Format(time.RFC3339Nano), "log_entry_id": ev.ID, "source_ip": ev.SourceIP, "remote_addr": ev.RemoteAddr, @@ -46,22 +48,26 @@ func BuildReplayRawEventPayload(ev models.LogEvent) (RawEventIngestBody, error) "device": ev.DeviceName, "job": "logs-replay", } - if uid := replayResourceUID(ev); uid != "" { - labels["resource_uid"] = uid - } + resourceUID := replayResourceUID(ev) + category, identity := splitResourceUID(resourceUID) + labels["resource_uid"] = resourceUID + labels["resource_category"] = category + labels["service_identity"] = identity return RawEventIngestBody{ - SourceType: sourceType, - ResourceUID: replayResourceUID(ev), - EventTime: time.Now().UTC(), - Severity: firstNonEmpty(ev.SeverityCode, "warning"), - Title: replayTitle(ev), - Message: firstNonEmpty(ev.NormalizedSummary, ev.RawPayload), - Labels: labels, + SourceType: sourceType, + SourceEventKey: fmt.Sprintf("logs:replay:%d:%d", ev.ID, now.UnixNano()), + State: "firing", + ResourceUID: resourceUID, + EventTime: now, + Severity: firstNonEmpty(ev.SeverityCode, "warning"), + Title: replayTitle(ev), + Message: firstNonEmpty(ev.NormalizedSummary, ev.RawPayload), + Labels: labels, Annotations: map[string]string{ "replay": "true", "dispatch_status": ev.DispatchStatus, }, - ParseStatus: "replayed", + ParseStatus: "parsed", RawPayload: rawBytes, }, nil } @@ -80,16 +86,42 @@ func EnqueueReplayLogEvent(ev models.LogEvent) (uint, error) { PayloadJSON: string(payload), Status: outboxStatusPending, RetryCount: 0, - NextRetryAt: time.Now(), } - if err := enqueueOutboxRow(&row); err != nil { + if impl.DBService == nil { + return 0, fmt.Errorf("database is not initialized") + } + if err := impl.DBService.Transaction(func(tx *gorm.DB) error { + now, err := databaseNow(tx) + if err != nil { + return err + } + row.NextRetryAt = now + if err := enqueueOutboxRowWithDB(tx, &row); err != nil { + return err + } + result := tx.Model(&models.LogEvent{}).Where("id = ?", ev.ID).Updates(map[string]interface{}{ + "dispatch_status": "pending", + "dispatch_outbox_id": row.ID, + "alert_sent": false, + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return fmt.Errorf("log event %d does not exist", ev.ID) + } + return nil + }); err != nil { return 0, err } return row.ID, nil } -func enqueueOutboxRow(row *models.AlertOutbox) error { - return impl.DBService.Create(row).Error +func enqueueOutboxRowWithDB(db *gorm.DB, row *models.AlertOutbox) error { + if db == nil { + return fmt.Errorf("database is not initialized") + } + return db.Create(row).Error } func replaySourceType(kind string) string { @@ -111,13 +143,32 @@ func replaySubtype(kind string) string { } func replayResourceUID(ev models.LogEvent) string { - if strings.Contains(ev.ResourceID, ":") { - return ev.ResourceID + if uid := strings.TrimSpace(ev.ResourceUID); uid != "" { + if category, identity, ok := splitKnownResourceUID(uid); ok { + return category + ":" + identity + } + return canonicalLogResourceCategory(ev.ResourceType, replaySubtype(ev.SourceKind)) + ":" + uid + } + if category, identity, ok := splitKnownResourceUID(ev.ResourceID); ok { + return category + ":" + identity } if ev.ResourceType != "" && ev.ResourceID != "" { - return ev.ResourceType + ":" + ev.ResourceID + return canonicalLogResourceCategory(ev.ResourceType, replaySubtype(ev.SourceKind)) + ":" + ev.ResourceID } - return "" + category := canonicalLogResourceCategory("", replaySubtype(ev.SourceKind)) + identity := firstNonEmpty(strings.TrimSpace(ev.SourceIP), strings.TrimSpace(ev.DeviceName)) + if identity == "" { + identity = fmt.Sprintf("log-event-%d", ev.ID) + } + return category + ":" + identity +} + +func splitResourceUID(uid string) (string, string) { + parts := strings.SplitN(strings.TrimSpace(uid), ":", 2) + if len(parts) != 2 { + return "log_source", strings.TrimSpace(uid) + } + return parts[0], parts[1] } func replayTitle(ev models.LogEvent) string { diff --git a/internal/logic/controllers/crud.go b/internal/logic/controllers/crud.go index 41b4ec8..f6fcbcb 100644 --- a/internal/logic/controllers/crud.go +++ b/internal/logic/controllers/crud.go @@ -344,10 +344,6 @@ func ReplayLogEvent(ctx *gin.Context) { infra.Response.Error(ctx, err) return } - if err := impl.DBService.Model(&models.LogEvent{}).Where("id = ?", id).Update("dispatch_status", "pending").Error; err != nil { - infra.Response.Error(ctx, err) - return - } infra.Response.Success(ctx, gin.H{ "log_event_id": id, "outbox_id": outboxID, diff --git a/internal/models/alert_outbox.go b/internal/models/alert_outbox.go index c8e5cc4..78c6300 100644 --- a/internal/models/alert_outbox.go +++ b/internal/models/alert_outbox.go @@ -1,29 +1,31 @@ -package models - -import "time" - -// AlertOutbox 表示待发送或重试中的告警任务。 -type AlertOutbox struct { - ID uint `gorm:"primaryKey" json:"id"` - - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` - - // LogEventID 关联日志事件 ID。 - LogEventID uint `gorm:"index" json:"log_event_id"` - // PayloadJSON 保存 AlertReceiveBody 的 JSON 文本。 - PayloadJSON string `gorm:"type:text" json:"payload_json"` - // Status 任务状态：pending/retrying/sent/dead。 - Status string `gorm:"size:32;index" json:"status"` - // RetryCount 已重试次数。 - RetryCount int `json:"retry_count"` - // NextRetryAt 下一次可重试时间。 - NextRetryAt time.Time `gorm:"index" json:"next_retry_at"` - // LastError 最近一次错误信息。 - LastError string `gorm:"type:text" json:"last_error"` -} - -func (AlertOutbox) TableName() string { - return "logs_alert_outbox" -} - +package models + +import "time" + +// AlertOutbox 表示待发送或重试中的告警任务。 +type AlertOutbox struct { + ID uint `gorm:"primaryKey;index:idx_logs_alert_outbox_event_latest,priority:2" json:"id"` + + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + + // LogEventID 关联日志事件 ID。 + LogEventID uint `gorm:"index;index:idx_logs_alert_outbox_event_latest,priority:1" json:"log_event_id"` + // PayloadJSON 保存 AlertReceiveBody 的 JSON 文本。 + PayloadJSON string `gorm:"type:text" json:"payload_json"` + // Status 任务状态：pending/processing/retrying/sent/dead。 + Status string `gorm:"size:32;index" json:"status"` + // RetryCount 已重试次数。 + RetryCount int `json:"retry_count"` + // NextRetryAt 下一次可重试时间。 + NextRetryAt time.Time `gorm:"index" json:"next_retry_at"` + // LeaseUntil/LeaseOwner 用于多实例原子领取，避免同一任务并发发送。 + LeaseUntil *time.Time `gorm:"index" json:"lease_until,omitempty"` + LeaseOwner string `gorm:"size:128;index" json:"lease_owner"` + // LastError 最近一次错误信息。 + LastError string `gorm:"type:text" json:"last_error"` +} + +func (AlertOutbox) TableName() string { + return "logs_alert_outbox" +} diff --git a/internal/models/log_event.go b/internal/models/log_event.go index 814bbea..978f7c2 100644 --- a/internal/models/log_event.go +++ b/internal/models/log_event.go @@ -1,45 +1,49 @@ -package models - -import "time" - -// LogEvent 表示一条归一化/存储后的日志事件。 -type LogEvent struct { - // ID 是数据库主键。 - ID uint `gorm:"primaryKey" json:"id"` - // CreatedAt 记录创建时间（写入日志事件时）。 - CreatedAt time.Time `json:"created_at"` - // SourceKind 表示日志来源类型（例如 trap/syslog 等）。 - SourceKind string `gorm:"size:16;index" json:"source_kind"` - // RemoteAddr 表示日志发送方地址。 - RemoteAddr string `gorm:"size:64" json:"remote_addr"` - // RawPayload 保存原始负载内容。 - RawPayload string `gorm:"type:text" json:"raw_payload"` - // NormalizedSummary 保存归一化后的摘要信息。 - NormalizedSummary string `gorm:"type:text" json:"normalized_summary"` - // NormalizedDetail 保存归一化后的详细信息。 - NormalizedDetail string `gorm:"type:text" json:"normalized_detail"` - // DeviceName 表示关联设备名称。 - DeviceName string `gorm:"size:512;index" json:"device_name"` - // SourceIP 表示原始来源 IP（不含端口）。 - SourceIP string `gorm:"size:64;index" json:"source_ip"` - // ResourceType 表示关联到的资源类型。 - ResourceType string `gorm:"size:32;index" json:"resource_type"` - // ResourceID 表示关联到的资源 ID。 - ResourceID string `gorm:"size:128;index" json:"resource_id"` - // ResourceName 表示关联到的资源名称。 - ResourceName string `gorm:"size:256" json:"resource_name"` - // MatchMethod 表示资源命中方式（ip/hostname/none）。 - MatchMethod string `gorm:"size:32" json:"match_method"` - // DispatchStatus 表示告警分发状态（not_applicable/pending/retrying/sent/dead）。 - DispatchStatus string `gorm:"size:32;index" json:"dispatch_status"` - // SeverityCode 表示告警/严重度编码。 - SeverityCode string `gorm:"size:32" json:"severity_code"` - // TrapOID 表示关联的 Trap OID（若来源为 trap）。 - TrapOID string `gorm:"size:512;index" json:"trap_oid"` - // AlertSent 表示是否已将告警发送出去。 - AlertSent bool `json:"alert_sent"` -} - -func (LogEvent) TableName() string { - return "logs_events" -} +package models + +import "time" + +// LogEvent 表示一条归一化/存储后的日志事件。 +type LogEvent struct { + // ID 是数据库主键。 + ID uint `gorm:"primaryKey" json:"id"` + // CreatedAt 记录创建时间（写入日志事件时）。 + CreatedAt time.Time `json:"created_at"` + // SourceKind 表示日志来源类型（例如 trap/syslog 等）。 + SourceKind string `gorm:"size:16;index" json:"source_kind"` + // RemoteAddr 表示日志发送方地址。 + RemoteAddr string `gorm:"size:64" json:"remote_addr"` + // RawPayload 保存原始负载内容。 + RawPayload string `gorm:"type:text" json:"raw_payload"` + // NormalizedSummary 保存归一化后的摘要信息。 + NormalizedSummary string `gorm:"type:text" json:"normalized_summary"` + // NormalizedDetail 保存归一化后的详细信息。 + NormalizedDetail string `gorm:"type:text" json:"normalized_detail"` + // DeviceName 表示关联设备名称。 + DeviceName string `gorm:"size:512;index" json:"device_name"` + // SourceIP 表示原始来源 IP（不含端口）。 + SourceIP string `gorm:"size:64;index" json:"source_ip"` + // ResourceType 表示关联到的资源类型。 + ResourceType string `gorm:"size:32;index" json:"resource_type"` + // ResourceUID 是跨服务使用的规范资源标识。 + ResourceUID string `gorm:"size:255;index" json:"resource_uid"` + // ResourceID 表示关联到的资源 ID。 + ResourceID string `gorm:"size:128;index" json:"resource_id"` + // ResourceName 表示关联到的资源名称。 + ResourceName string `gorm:"size:256" json:"resource_name"` + // MatchMethod 表示资源命中方式（ip/hostname/none）。 + MatchMethod string `gorm:"size:32" json:"match_method"` + // DispatchStatus 表示告警分发状态（not_applicable/pending/retrying/sent/dead）。 + DispatchStatus string `gorm:"size:32;index" json:"dispatch_status"` + // DispatchOutboxID 标识当前一次分发，防止旧任务覆盖较新的重放状态。 + DispatchOutboxID uint `gorm:"not null;default:0;index" json:"dispatch_outbox_id"` + // SeverityCode 表示告警/严重度编码。 + SeverityCode string `gorm:"size:32" json:"severity_code"` + // TrapOID 表示关联的 Trap OID（若来源为 trap）。 + TrapOID string `gorm:"size:512;index" json:"trap_oid"` + // AlertSent 表示是否已将告警发送出去。 + AlertSent bool `json:"alert_sent"` +} + +func (LogEvent) TableName() string { + return "logs_events" +} diff --git a/internal/models/query.go b/internal/models/query.go index ee6a287..e70da3e 100644 --- a/internal/models/query.go +++ b/internal/models/query.go @@ -23,6 +23,9 @@ func InitData(db *gorm.DB) error { if db == nil { return nil } + if err := backfillDispatchOutboxIDs(db); err != nil { + return err + } if err := seedDefaultSyslogRules(db); err != nil { return err } @@ -35,6 +38,52 @@ func InitData(db *gorm.DB) error { return nil } +func backfillDispatchOutboxIDs(db *gorm.DB) error { + const postgresUpdate = ` +WITH latest AS ( + SELECT target_event.id AS log_event_id, current_outbox.id, current_outbox.status + FROM logs_events AS target_event + JOIN LATERAL ( + SELECT id, status + FROM logs_alert_outbox + WHERE log_event_id = target_event.id + ORDER BY id DESC + LIMIT 1 + ) AS current_outbox ON TRUE + WHERE target_event.dispatch_outbox_id = 0 +) +UPDATE logs_events AS target_event +SET dispatch_outbox_id = latest.id, + dispatch_status = CASE latest.status WHEN 'processing' THEN 'retrying' ELSE latest.status END, + alert_sent = CASE WHEN latest.status = 'sent' THEN TRUE ELSE FALSE END +FROM latest +WHERE target_event.id = latest.log_event_id + AND target_event.dispatch_outbox_id = 0` + const mysqlUpdate = ` +UPDATE logs_events AS target_event +JOIN ( + SELECT candidate.id AS log_event_id, MAX(outbox.id) AS id + FROM logs_events AS candidate + JOIN logs_alert_outbox AS outbox ON outbox.log_event_id = candidate.id + WHERE candidate.dispatch_outbox_id = 0 + GROUP BY candidate.id +) AS latest_id ON latest_id.log_event_id = target_event.id +JOIN logs_alert_outbox AS latest ON latest.id = latest_id.id +SET dispatch_outbox_id = latest.id, + dispatch_status = CASE latest.status WHEN 'processing' THEN 'retrying' ELSE latest.status END, + alert_sent = CASE WHEN latest.status = 'sent' THEN TRUE ELSE FALSE END +WHERE target_event.dispatch_outbox_id = 0` + + switch db.Dialector.Name() { + case "postgres": + return db.Exec(postgresUpdate).Error + case "mysql": + return db.Exec(mysqlUpdate).Error + default: + return gorm.ErrUnsupportedDriver + } +} + func seedDefaultSyslogRules(db *gorm.DB) error { var cnt int64 if err := db.Model(&SyslogRule{}).Count(&cnt).Error; err != nil {