ops/logs
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: make log alert delivery durable

Browse Source
This commit is contained in:
zxr
2026-07-21 02:32:18 +08:00
parent ce559d5218
commit 6e66bad884
8 changed files with 709 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
internal/ingest/alert_forward.go
View File

@@ -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"]) {

270
internal/ingest/alert_outbox.go
View File

@@ -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])
}

305
internal/ingest/engine.go
View File

@@ -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
}
}

91
internal/ingest/replay.go
View File

@@ -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 {