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

fix: correlate log alert recovery lifecycle

Browse Source
This commit is contained in:
zxr
2026-07-21 22:15:59 +08:00
parent eaca3d0ac8
commit 8495fdf039
6 changed files with 194 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
internal/ingest/engine.go
View File

@@ -1,6 +1,8 @@
package ingest
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log"
@@ -267,16 +269,13 @@ func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) {
if err != nil {
return 0, err
}
state := "firing"
if isRecoverySignal(parsed.Message, parsed.RawLine) {
state = "resolved"
}
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: state, SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes,
State: matchDetails.State, SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes,
}
body.Fingerprint = alertLifecycleFingerprint(body, matched.LifecycleKey, "")
return enqueueAlertWithDB(tx, stored.ID, body)
}); err != nil {
log.Printf("logs: persist matched syslog event: %v", err)
@@ -285,6 +284,7 @@ func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) {
type syslogRuleMatch struct {
Matched bool
State string
ResourceUID string
SeverityCode string
Captures map[string]string
@@ -295,11 +295,12 @@ func syslogRuleMatches(rule *models.SyslogRule, device, message, rawLine string)
}
func syslogRuleMatchDetails(rule *models.SyslogRule, device, message, rawLine string) syslogRuleMatch {
result := syslogRuleMatch{Captures: map[string]string{}}
result := syslogRuleMatch{State: "firing", Captures: map[string]string{}}
deviceContains := strings.TrimSpace(rule.DeviceNameContains)
sourceMatch := strings.TrimSpace(rule.SourceMatch)
keywordRegex := strings.TrimSpace(rule.KeywordRegex)
messageRegex := strings.TrimSpace(rule.MessageRegex)
recoveryRegex := strings.TrimSpace(rule.RecoveryMatchRegex)
if deviceContains == "" && sourceMatch == "" && keywordRegex == "" && messageRegex == "" {
return result
}
@@ -316,24 +317,35 @@ func syslogRuleMatchDetails(rule *models.SyslogRule, device, message, rawLine st
return result
}
}
for _, pattern := range []string{keywordRegex, messageRegex} {
if pattern == "" {
continue
}
re, err := regexp.Compile(pattern)
if recoveryRegex != "" {
re, err := regexp.Compile(recoveryRegex)
if err != nil {
return result
}
matches := re.FindStringSubmatch(message)
if matches == nil {
matches = re.FindStringSubmatch(rawLine)
matches := firstRegexMatch(re, message, rawLine)
if matches != nil {
mergeNamedCaptures(result.Captures, re, matches)
result.State = "resolved"
result.Matched = true
}
if matches == nil {
return result
}
mergeNamedCaptures(result.Captures, re, matches)
}
result.Matched = true
if !result.Matched {
for _, pattern := range []string{keywordRegex, messageRegex} {
if pattern == "" {
continue
}
re, err := regexp.Compile(pattern)
if err != nil {
return result
}
matches := firstRegexMatch(re, message, rawLine)
if matches == nil {
return result
}
mergeNamedCaptures(result.Captures, re, matches)
}
result.Matched = true
}
if uid := extractWithNamedRegex(rule.ResourceUIDExtractRegex, "resource_uid", message, rawLine); uid != "" {
result.ResourceUID = normalizeExtractedResourceUID(uid)
} else if uid := result.Captures["resource_uid"]; uid != "" {
@@ -343,6 +355,15 @@ func syslogRuleMatchDetails(rule *models.SyslogRule, device, message, rawLine st
return result
}
func firstRegexMatch(re *regexp.Regexp, values ...string) []string {
for _, value := range values {
if matches := re.FindStringSubmatch(value); matches != nil {
return matches
}
}
return nil
}
func mergeNamedCaptures(dst map[string]string, re *regexp.Regexp, matches []string) {
names := re.SubexpNames()
for i, name := range names {
@@ -370,7 +391,9 @@ func extractWithNamedRegex(pattern, groupName, message, rawLine string) string {
names := re.SubexpNames()
for i, name := range names {
if i > 0 && name == groupName && i < len(matches) {
return strings.TrimSpace(matches[i])
if value := strings.TrimSpace(matches[i]); value != "" {
return value
}
}
}
for i := 1; i < len(matches); i++ {
@@ -510,8 +533,8 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
rules := e.trapRules
e.mu.RUnlock()
matched := firstMatchingTrapRule(rules, trapOID, fp)
if matched == nil {
match := firstMatchingTrapRule(rules, trapOID, fp)
if match.Rule == nil {
rawBytes, mErr := json.Marshal(fp)
if mErr != nil {
return
@@ -541,6 +564,7 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
}
return
}
matched := match.Rule
desc := readable
if dict != nil && dict.RecoveryMessage != "" {
@@ -555,6 +579,10 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
"instance": addr.IP.String(),
"job": "logs-trap",
}
trapInstance := trapInstanceKey(pkt)
if trapInstance != "" {
labels["trap_instance"] = trapInstance
}
if matched.ID != 0 {
labels["resource_type"] = "trap_rule"
labels["resource_id"] = strconv.FormatUint(uint64(matched.ID), 10)
@@ -588,20 +616,13 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
if err != nil {
return 0, err
}
state := "firing"
dictText := ""
if dict != nil {
dictText = firstNonEmpty(dict.Name, dict.Title)
}
if isRecoverySignal(readable, dictText, matched.Name, matched.AlertName) {
state = "resolved"
}
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: state,
Agent: "logs-trap", PolicyID: matched.PolicyID, State: match.State,
SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes,
}
body.Fingerprint = alertLifecycleFingerprint(body, matched.LifecycleKey, trapInstance)
return enqueueAlertWithDB(tx, stored.ID, body)
}); err != nil {
log.Printf("logs: persist matched trap event: %v", err)
@@ -646,6 +667,27 @@ func trapVarbinds(pkt *gosnmp.SnmpPacket) []map[string]string {
return out
}
func trapInstanceKey(pkt *gosnmp.SnmpPacket) string {
if pkt == nil {
return ""
}
for _, prefix := range []string{
"1.3.6.1.2.1.31.1.1.1.1.",
"1.3.6.1.2.1.2.2.1.2.",
"1.3.6.1.2.1.2.2.1.1.",
} {
for _, variable := range pkt.Variables {
name := normOID(variable.Name)
if strings.HasPrefix(name, prefix) {
if index := strings.TrimSpace(strings.TrimPrefix(name, prefix)); index != "" {
return index
}
}
}
}
return ""
}
func buildTrapReadable(trapOID string, dict *models.TrapDictionaryEntry, varbindSummary string) string {
if dict != nil && firstNonEmpty(dict.Name, dict.Title) != "" {
return firstNonEmpty(dict.Name, dict.Title) + " (" + trapOID + ")"
@@ -656,34 +698,49 @@ func buildTrapReadable(trapOID string, dict *models.TrapDictionaryEntry, varbind
return truncate(varbindSummary, 256)
}
func trapRuleMatches(rule *models.TrapRule, trapOID, varbindFP string) bool {
type trapRuleMatch struct {
Rule *models.TrapRule
State string
}
func trapRuleState(rule *models.TrapRule, trapOID, varbindFP string) (string, bool) {
hasOID := strings.TrimSpace(rule.OIDPrefix) != ""
hasRE := strings.TrimSpace(rule.VarbindMatchRegex) != ""
hasRecoveryRE := strings.TrimSpace(rule.RecoveryMatchRegex) != ""
if !hasOID && !hasRE {
return false
return "", false
}
if hasOID && !strings.HasPrefix(normOID(trapOID), normOID(rule.OIDPrefix)) {
return false
return "", false
}
if hasRecoveryRE {
re, err := regexp.Compile(rule.RecoveryMatchRegex)
if err != nil {
return "", false
}
if re.MatchString(trapOID) || re.MatchString(varbindFP) {
return "resolved", true
}
}
if hasRE {
re, err := regexp.Compile(rule.VarbindMatchRegex)
if err != nil {
return false
return "", false
}
if !re.MatchString(varbindFP) {
return false
return "", false
}
}
return true
return "firing", true
}
func firstMatchingTrapRule(rules []models.TrapRule, trapOID, varbindFP string) *models.TrapRule {
func firstMatchingTrapRule(rules []models.TrapRule, trapOID, varbindFP string) trapRuleMatch {
for i := range rules {
if trapRuleMatches(&rules[i], trapOID, varbindFP) {
return &rules[i]
if state, matched := trapRuleState(&rules[i], trapOID, varbindFP); matched {
return trapRuleMatch{Rule: &rules[i], State: state}
}
}
return nil
return trapRuleMatch{}
}
func firstNonEmpty(a, b string) string {
@@ -693,19 +750,27 @@ func firstNonEmpty(a, b string) string {
return b
}
func isRecoverySignal(values ...string) bool {
for _, value := range values {
text := strings.ToLower(strings.TrimSpace(value))
if text == "" {
continue
}
for _, marker := range []string{"恢复", "recovered", "recovery", "ifup", "link up", "port up", "interface up", "normal"} {
if strings.Contains(text, marker) {
return true
}
}
func alertLifecycleFingerprint(body AlertReceiveBody, lifecycleKey, instance string) string {
lifecycleKey = strings.TrimSpace(lifecycleKey)
if lifecycleKey == "" {
return ""
}
return false
resourceUID := ""
sourceIP := ""
if body.Labels != nil {
resourceUID = strings.TrimSpace(body.Labels["resource_uid"])
sourceIP = strings.TrimSpace(body.Labels["ip"])
}
identity := strings.Join([]string{
strings.TrimSpace(body.Agent),
resourceUID,
sourceIP,
strconv.FormatUint(uint64(body.PolicyID), 10),
lifecycleKey,
strings.TrimSpace(instance),
}, "\x00")
sum := sha256.Sum256([]byte(identity))
return hex.EncodeToString(sum[:])
}
func (e *Engine) resolveResource(sourceIP, hostname string) (resourceRef, string) {