From 972adb3d38d7bc7fe93b3bbba147f88b92abff79 Mon Sep 17 00:00:00 2001 From: zxr <271055687@qq.com> Date: Mon, 21 Sep 2026 10:46:34 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=AA=8C=E6=94=B6=E4=BF=AE=E6=AD=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/ingest/engine.go | 12 +- internal/ingest/shield.go | 260 ++++++++++-------- internal/logic/controllers/crud.go | 8 + internal/logic/controllers/rule_validation.go | 53 ++++ .../logic/controllers/trap_shield_options.go | 82 ++++++ internal/logic/otlp/ingest.go | 9 +- internal/logic/otlp/query.go | 112 ++++++++ internal/models/log_event.go | 2 + internal/models/schema_version.go | 58 +++- internal/models/trap_shield.go | 58 ++-- internal/routers/register.go | 2 + scripts/run_logs_e2e.py | 18 +- 12 files changed, 506 insertions(+), 168 deletions(-) create mode 100644 internal/logic/controllers/trap_shield_options.go diff --git a/internal/ingest/engine.go b/internal/ingest/engine.go index 61f0de1..9c15e45 100644 --- a/internal/ingest/engine.go +++ b/internal/ingest/engine.go @@ -463,7 +463,7 @@ func trapShielded(e *Engine, addr *net.UDPAddr, trapOID string, pkt *gosnmp.Snmp if cidr := strings.TrimSpace(s.SourceIPCIDR); cidr != "" && !ipMatchesCIDR(ip, cidr) { continue } - if p := strings.TrimSpace(s.OIDPrefix); p != "" && !strings.HasPrefix(normOID(trapOID), normOID(p)) { + if len(s.OIDPrefixes) > 0 && !matchesTrapOIDPrefix(trapOID, s.OIDPrefixes) { continue } if h := strings.TrimSpace(s.InterfaceHint); h != "" && !strings.Contains(fp, h) { @@ -477,6 +477,16 @@ func trapShielded(e *Engine, addr *net.UDPAddr, trapOID string, pkt *gosnmp.Snmp return false } +func matchesTrapOIDPrefix(trapOID string, prefixes []string) bool { + normalizedOID := normOID(trapOID) + for _, prefix := range prefixes { + if normalizedPrefix := normOID(prefix); normalizedPrefix != "" && strings.HasPrefix(normalizedOID, normalizedPrefix) { + return true + } + } + return false +} + func lookupTrapDict(e *Engine, trapOID string) *models.TrapDictionaryEntry { t := normOID(trapOID) e.mu.RLock() diff --git a/internal/ingest/shield.go b/internal/ingest/shield.go index 59a6142..275f975 100644 --- a/internal/ingest/shield.go +++ b/internal/ingest/shield.go @@ -1,114 +1,146 @@ -package ingest - -import ( - "encoding/json" - "fmt" - "net" - "strconv" - "strings" - "time" - - "github.com/gosnmp/gosnmp" -) - -type timeWindow struct { - Days []int `json:"days"` - Start string `json:"start"` - End string `json:"end"` -} - -func ipMatchesCIDR(ip net.IP, cidr string) bool { - cidr = strings.TrimSpace(cidr) - if cidr == "" { - return false - } - if !strings.Contains(cidr, "/") { - p := net.ParseIP(cidr) - return p != nil && p.Equal(ip) - } - _, network, err := net.ParseCIDR(cidr) - if err != nil { - return false - } - return network.Contains(ip) -} - -func inTimeWindows(now time.Time, jsonStr string) bool { - s := strings.TrimSpace(jsonStr) - if s == "" || s == "null" { - return true - } - var windows []timeWindow - if err := json.Unmarshal([]byte(s), &windows); err != nil || len(windows) == 0 { - return false - } - tod := now.Hour()*60 + now.Minute() - wd := int(now.Weekday()) - for _, w := range windows { - if len(w.Days) > 0 { - ok := false - for _, d := range w.Days { - if d == wd { - ok = true - break - } - } - if !ok { - continue - } - } - start := parseHHMM(w.Start) - end := parseHHMM(w.End) - if start < 0 || end < 0 { - continue - } - if start <= end { - if tod >= start && tod <= end { - return true - } - } else { - if tod >= start || tod <= end { - return true - } - } - } - return false -} - -func parseHHMM(s string) int { - s = strings.TrimSpace(s) - if s == "" { - return -1 - } - parts := strings.Split(s, ":") - if len(parts) != 2 { - return -1 - } - h, err1 := strconv.Atoi(parts[0]) - m, err2 := strconv.Atoi(parts[1]) - if err1 != nil || err2 != nil || h < 0 || h > 23 || m < 0 || m > 59 { - return -1 - } - return h*60 + m -} - -func varbindFingerprint(pkt *gosnmp.SnmpPacket) string { - var b strings.Builder - for _, v := range pkt.Variables { - b.WriteString(v.Name) - b.WriteByte('=') - b.WriteString(fmtVarbindValue(v)) - b.WriteByte(';') - } - return b.String() -} - -func fmtVarbindValue(v gosnmp.SnmpPDU) string { - switch v.Type { - case gosnmp.OctetString: - if bb, ok := v.Value.([]byte); ok { - return string(bb) - } - } - return fmt.Sprintf("%v", v.Value) -} +package ingest + +import ( + "encoding/json" + "fmt" + "net" + "strconv" + "strings" + "time" + + "github.com/gosnmp/gosnmp" +) + +type timeWindow struct { + Days []int `json:"days"` + Start string `json:"start"` + End string `json:"end"` +} + +// ValidateTimeWindowsJSON 校验 Trap 屏蔽规则的生效时间窗。 +// 空字符串表示始终生效；非空时必须是至少包含一个有效时间段的 JSON 数组。 +func ValidateTimeWindowsJSON(jsonStr string) error { + s := strings.TrimSpace(jsonStr) + if s == "" { + return nil + } + var windows []timeWindow + if err := json.Unmarshal([]byte(s), &windows); err != nil { + return fmt.Errorf("时间窗必须是 JSON 数组：%w", err) + } + if len(windows) == 0 { + return fmt.Errorf("时间窗不能为空数组；不限制时间时请清空该字段") + } + for i, window := range windows { + if parseHHMM(window.Start) < 0 || parseHHMM(window.End) < 0 { + return fmt.Errorf("第 %d 个时间窗的开始或结束时间无效，应使用 HH:MM 格式", i+1) + } + seenDays := make(map[int]struct{}, len(window.Days)) + for _, day := range window.Days { + if day < 0 || day > 6 { + return fmt.Errorf("第 %d 个时间窗的星期值无效，应为 0 到 6", i+1) + } + if _, exists := seenDays[day]; exists { + return fmt.Errorf("第 %d 个时间窗包含重复的星期值 %d", i+1, day) + } + seenDays[day] = struct{}{} + } + } + return nil +} + +func ipMatchesCIDR(ip net.IP, cidr string) bool { + cidr = strings.TrimSpace(cidr) + if cidr == "" { + return false + } + if !strings.Contains(cidr, "/") { + p := net.ParseIP(cidr) + return p != nil && p.Equal(ip) + } + _, network, err := net.ParseCIDR(cidr) + if err != nil { + return false + } + return network.Contains(ip) +} + +func inTimeWindows(now time.Time, jsonStr string) bool { + s := strings.TrimSpace(jsonStr) + if s == "" { + return true + } + var windows []timeWindow + if err := json.Unmarshal([]byte(s), &windows); err != nil || len(windows) == 0 { + return false + } + tod := now.Hour()*60 + now.Minute() + wd := int(now.Weekday()) + for _, w := range windows { + if len(w.Days) > 0 { + ok := false + for _, d := range w.Days { + if d == wd { + ok = true + break + } + } + if !ok { + continue + } + } + start := parseHHMM(w.Start) + end := parseHHMM(w.End) + if start < 0 || end < 0 { + continue + } + if start <= end { + if tod >= start && tod <= end { + return true + } + } else { + if tod >= start || tod <= end { + return true + } + } + } + return false +} + +func parseHHMM(s string) int { + s = strings.TrimSpace(s) + if s == "" { + return -1 + } + parts := strings.Split(s, ":") + if len(parts) != 2 { + return -1 + } + h, err1 := strconv.Atoi(parts[0]) + m, err2 := strconv.Atoi(parts[1]) + if err1 != nil || err2 != nil || h < 0 || h > 23 || m < 0 || m > 59 { + return -1 + } + return h*60 + m +} + +func varbindFingerprint(pkt *gosnmp.SnmpPacket) string { + var b strings.Builder + for _, v := range pkt.Variables { + b.WriteString(v.Name) + b.WriteByte('=') + b.WriteString(fmtVarbindValue(v)) + b.WriteByte(';') + } + return b.String() +} + +func fmtVarbindValue(v gosnmp.SnmpPDU) string { + switch v.Type { + case gosnmp.OctetString: + if bb, ok := v.Value.([]byte); ok { + return string(bb) + } + } + return fmt.Sprintf("%v", v.Value) +} diff --git a/internal/logic/controllers/crud.go b/internal/logic/controllers/crud.go index 43ffb20..0431840 100644 --- a/internal/logic/controllers/crud.go +++ b/internal/logic/controllers/crud.go @@ -241,6 +241,10 @@ func CreateTrapShield(ctx *gin.Context) { infra.Response.Error(ctx, err) return } + if err := validateTrapShield(&row); err != nil { + infra.Response.Error(ctx, err) + return + } row.ID = 0 if err := impl.DBService.Create(&row).Error; err != nil { infra.Response.Error(ctx, err) @@ -266,6 +270,10 @@ func UpdateTrapShield(ctx *gin.Context) { return } row.ID = id + if err := validateTrapShield(&row); err != nil { + infra.Response.Error(ctx, err) + return + } if err := impl.DBService.Save(&row).Error; err != nil { infra.Response.Error(ctx, err) return diff --git a/internal/logic/controllers/rule_validation.go b/internal/logic/controllers/rule_validation.go index 385c373..befe819 100644 --- a/internal/logic/controllers/rule_validation.go +++ b/internal/logic/controllers/rule_validation.go @@ -3,12 +3,16 @@ package controllers import ( "encoding/json" "fmt" + "net" "regexp" "strings" + "git.apinb.com/ops/logs/internal/ingest" "git.apinb.com/ops/logs/internal/models" ) +var trapOIDPattern = regexp.MustCompile(`^\d+(?:\.\d+)*$`) + func validateSyslogRule(rule *models.SyslogRule) error { regexFields := []struct { name string @@ -71,6 +75,55 @@ func validateTrapRule(rule *models.TrapRule) error { return nil } +func validateTrapShield(shield *models.TrapShield) error { + shield.Name = strings.TrimSpace(shield.Name) + shield.SourceIPCIDR = strings.TrimSpace(shield.SourceIPCIDR) + shield.InterfaceHint = strings.TrimSpace(shield.InterfaceHint) + shield.TimeWindowsJSON = strings.TrimSpace(shield.TimeWindowsJSON) + if shield.Name == "" { + return fmt.Errorf("Trap 屏蔽规则名称不能为空") + } + if shield.SourceIPCIDR != "" { + if strings.Contains(shield.SourceIPCIDR, "/") { + if _, _, err := net.ParseCIDR(shield.SourceIPCIDR); err != nil { + return fmt.Errorf("源 IP / CIDR 格式无效：%v", err) + } + } else if net.ParseIP(shield.SourceIPCIDR) == nil { + return fmt.Errorf("源 IP / CIDR 格式无效") + } + } + + normalizedPrefixes := make([]string, 0, len(shield.OIDPrefixes)) + seenPrefixes := make(map[string]struct{}, len(shield.OIDPrefixes)) + for _, raw := range shield.OIDPrefixes { + prefix := normalizeTrapOIDInput(raw) + if prefix == "" { + continue + } + if !trapOIDPattern.MatchString(prefix) { + return fmt.Errorf("Trap OID %q 格式无效", raw) + } + if _, exists := seenPrefixes[prefix]; exists { + continue + } + seenPrefixes[prefix] = struct{}{} + normalizedPrefixes = append(normalizedPrefixes, prefix) + } + shield.OIDPrefixes = normalizedPrefixes + + if shield.SourceIPCIDR == "" && len(shield.OIDPrefixes) == 0 && shield.InterfaceHint == "" { + return fmt.Errorf("Trap 屏蔽范围不能为空；请至少设置源 IP / CIDR、Trap 类型或接口提示中的一项") + } + if err := ingest.ValidateTimeWindowsJSON(shield.TimeWindowsJSON); err != nil { + return err + } + return nil +} + +func normalizeTrapOIDInput(value string) string { + return strings.Trim(strings.TrimSpace(value), ".") +} + func validateOptionalRegex(field, pattern string) error { pattern = strings.TrimSpace(pattern) if pattern == "" { diff --git a/internal/logic/controllers/trap_shield_options.go b/internal/logic/controllers/trap_shield_options.go new file mode 100644 index 0000000..a15e351 --- /dev/null +++ b/internal/logic/controllers/trap_shield_options.go @@ -0,0 +1,82 @@ +package controllers + +import ( + "strings" + + "git.apinb.com/bsm-sdk/core/infra" + "git.apinb.com/ops/logs/internal/impl" + "git.apinb.com/ops/logs/internal/models" + "github.com/gin-gonic/gin" +) + +type trapShieldOption struct { + OIDPrefix string `json:"oid_prefix"` + Name string `json:"name"` + Vendor string `json:"vendor"` + Source string `json:"source"` +} + +// ListTrapShieldOptions 返回 Trap 字典和已接收 Trap 中可用于屏蔽配置的类型。 +func ListTrapShieldOptions(ctx *gin.Context) { + var dictionary []models.TrapDictionaryEntry + if err := impl.DBService.Order("vendor asc, name asc, id asc").Find(&dictionary).Error; err != nil { + infra.Response.Error(ctx, err) + return + } + + options := make([]trapShieldOption, 0, len(dictionary)) + seen := make(map[string]struct{}, len(dictionary)) + for _, entry := range dictionary { + oid := normalizeTrapOIDInput(entry.OID) + if oid == "" { + oid = normalizeTrapOIDInput(entry.OIDPrefix) + } + if oid == "" { + continue + } + if _, exists := seen[oid]; exists { + continue + } + seen[oid] = struct{}{} + name := strings.TrimSpace(entry.Name) + if name == "" { + name = strings.TrimSpace(entry.Title) + } + if name == "" { + name = oid + } + options = append(options, trapShieldOption{ + OIDPrefix: oid, + Name: name, + Vendor: strings.TrimSpace(entry.Vendor), + Source: "dictionary", + }) + } + + var receivedOIDs []string + if err := impl.DBService.Model(&models.LogEvent{}). + Where("source_kind = ? AND trap_o_id <> ?", "snmp_trap", ""). + Distinct("trap_o_id"). + Order("trap_o_id asc"). + Pluck("trap_o_id", &receivedOIDs).Error; err != nil { + infra.Response.Error(ctx, err) + return + } + for _, rawOID := range receivedOIDs { + oid := normalizeTrapOIDInput(rawOID) + if oid == "" { + continue + } + if _, exists := seen[oid]; exists { + continue + } + seen[oid] = struct{}{} + options = append(options, trapShieldOption{ + OIDPrefix: oid, + Name: "已接收 Trap", + Source: "received", + }) + } + + infra.Response.Success(ctx, gin.H{"items": options}) +} diff --git a/internal/logic/otlp/ingest.go b/internal/logic/otlp/ingest.go index ea49479..14f0eda 100644 --- a/internal/logic/otlp/ingest.go +++ b/internal/logic/otlp/ingest.go @@ -128,7 +128,8 @@ func ReceiveLogs(ctx *gin.Context) { } rows, rejected, message := decodeLogs(request) if len(rows) > 0 { - if err := impl.DBService.CreateInBatches(rows, 500).Error; err != nil { + if err := impl.DBService.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "ingest_key"}}, DoNothing: true}). + CreateInBatches(rows, 500).Error; err != nil { ctx.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{"error": "Log 入库失败"}) return } @@ -282,8 +283,12 @@ func logModel(input logPayload, resourceAttrs map[string]interface{}) (models.Lo } serviceName := limitCharacters(textAttribute(resourceAttrs, "service.name"), 512) resourceUID := limitCharacters(firstTextAttribute(resourceAttrs, attrs, "resource.uid", "resource_uid"), 255) + var ingestKey *string + if value := limitCharacters(firstTextAttribute(attrs, nil, "log.record.uid"), 160); value != "" { + ingestKey = &value + } return models.LogEvent{ - CreatedAt: timeFromLog(input), SourceKind: "otlp", RawPayload: body, NormalizedSummary: body, + IngestKey: ingestKey, CreatedAt: timeFromLog(input), SourceKind: "otlp", RawPayload: body, NormalizedSummary: body, NormalizedDetail: string(detailJSON), DeviceName: serviceName, ResourceType: "service", ResourceUID: resourceUID, ResourceID: resourceUID, ResourceName: serviceName, BusinessSystemID: uintAttribute(resourceAttrs, attrs, "business.system.id", "business_system_id"), TraceID: traceID, SpanID: spanID, MatchMethod: "otlp", DispatchStatus: "not_applicable", diff --git a/internal/logic/otlp/query.go b/internal/logic/otlp/query.go index 6f27159..f7c6a72 100644 --- a/internal/logic/otlp/query.go +++ b/internal/logic/otlp/query.go @@ -1,8 +1,11 @@ package otlp import ( + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" + "fmt" "sort" "strconv" "strings" @@ -41,6 +44,19 @@ type dependencyAggregate struct { DurationMs float64 } +type middlewareCandidate struct { + CandidateKey string `json:"candidate_key"` + Product string `json:"product"` + DisplayName string `json:"display_name"` + Address string `json:"address,omitempty"` + SourceService string `json:"source_service"` + BusinessSystemID uint `json:"business_system_id"` + FirstSeenAt time.Time `json:"first_seen_at"` + LastSeenAt time.Time `json:"last_seen_at"` + SpanCount int64 `json:"span_count"` + ErrorCount int64 `json:"error_count"` +} + func ListTraces(ctx *gin.Context) { page, size := pageParams(ctx) query := applyTraceFilters(impl.DBService.Model(&models.TraceSpan{}), ctx) @@ -128,6 +144,102 @@ func ListDependencies(ctx *gin.Context) { infra.Response.Success(ctx, gin.H{"items": edges, "count": len(edges)}) } +// ListMiddlewareCandidates 从调用链标准属性中识别尚待管理员确认的中间件候选，不自动创建设备。 +func ListMiddlewareCandidates(ctx *gin.Context) { + businessSystemID, err := strconv.ParseUint(strings.TrimSpace(ctx.Query("business_system_id")), 10, 32) + if err != nil || businessSystemID == 0 { + infra.Response.Error(ctx, errors.New("business_system_id 无效")) + return + } + start := time.Now().UTC().Add(-30 * 24 * time.Hour) + if value, parseErr := time.Parse(time.RFC3339, strings.TrimSpace(ctx.Query("start_time"))); parseErr == nil { + start = value.UTC() + } + var rows []models.TraceSpan + if err := impl.DBService.Where("business_system_id = ? AND start_time >= ?", uint(businessSystemID), start). + Order("start_time DESC").Limit(100000).Find(&rows).Error; err != nil { + infra.Response.Error(ctx, err) + return + } + aggregates := make(map[string]*middlewareCandidate) + for _, row := range rows { + product, displayName, address, ok := middlewareIdentity(row) + if !ok { + continue + } + key := middlewareCandidateKey(uint(businessSystemID), product, displayName, address) + item := aggregates[key] + if item == nil { + item = &middlewareCandidate{ + CandidateKey: key, Product: product, DisplayName: displayName, Address: address, + SourceService: row.ServiceName, BusinessSystemID: uint(businessSystemID), + FirstSeenAt: row.StartTime, LastSeenAt: row.StartTime, + } + aggregates[key] = item + } + item.SpanCount++ + if row.StatusCode == "error" { + item.ErrorCount++ + } + if row.StartTime.Before(item.FirstSeenAt) { + item.FirstSeenAt = row.StartTime + } + if row.StartTime.After(item.LastSeenAt) { + item.LastSeenAt = row.StartTime + item.SourceService = row.ServiceName + } + } + items := make([]middlewareCandidate, 0, len(aggregates)) + for _, item := range aggregates { + items = append(items, *item) + } + sort.Slice(items, func(left, right int) bool { + if !items[left].LastSeenAt.Equal(items[right].LastSeenAt) { + return items[left].LastSeenAt.After(items[right].LastSeenAt) + } + return items[left].CandidateKey < items[right].CandidateKey + }) + infra.Response.Success(ctx, gin.H{"items": items, "count": len(items), "start_time": start}) +} + +func middlewareIdentity(row models.TraceSpan) (string, string, string, bool) { + attrs := make(map[string]interface{}) + resourceAttrs := make(map[string]interface{}) + _ = json.Unmarshal([]byte(row.Attributes), &attrs) + _ = json.Unmarshal([]byte(row.ResourceAttrs), &resourceAttrs) + rawProduct := firstTextAttribute(attrs, resourceAttrs, "messaging.system", "db.system.name", "db.system", "peer.service") + product := knownMiddlewareProduct(rawProduct) + if product == "" { + return "", "", "", false + } + displayName := firstTextAttribute(attrs, resourceAttrs, "peer.service", "server.address", "network.peer.address", "net.peer.name") + if displayName == "" { + displayName = product + } + address := firstTextAttribute(attrs, resourceAttrs, "server.address", "network.peer.address", "net.peer.name") + port := firstTextAttribute(attrs, resourceAttrs, "server.port", "network.peer.port", "net.peer.port") + if address != "" && port != "" && !strings.Contains(address, ":") { + address += ":" + port + } + return product, limitCharacters(displayName, 255), limitCharacters(address, 255), true +} + +func knownMiddlewareProduct(value string) string { + value = strings.ToLower(strings.TrimSpace(value)) + for _, product := range []string{"nginx", "apache", "tomcat", "redis", "kafka", "rabbitmq", "elasticsearch", "activemq", "rocketmq", "websphere"} { + if strings.Contains(value, product) { + return product + } + } + return "" +} + +func middlewareCandidateKey(businessSystemID uint, product, name, address string) string { + raw := fmt.Sprintf("%d\x00%s\x00%s\x00%s", businessSystemID, product, strings.ToLower(name), strings.ToLower(address)) + digest := sha256.Sum256([]byte(raw)) + return hex.EncodeToString(digest[:]) +} + func addDependency(aggregates map[string]*dependencyAggregate, source, target string, span models.TraceSpan) { key := source + "\x00" + target item := aggregates[key] diff --git a/internal/models/log_event.go b/internal/models/log_event.go index a3ae056..372a979 100644 --- a/internal/models/log_event.go +++ b/internal/models/log_event.go @@ -6,6 +6,8 @@ import "time" type LogEvent struct { // ID 是数据库主键。 ID uint `gorm:"primaryKey" json:"id"` + // IngestKey 是主动文件日志等可重试来源提供的稳定幂等键。 + IngestKey *string `gorm:"size:160;uniqueIndex" json:"ingest_key,omitempty"` // CreatedAt 记录创建时间（写入日志事件时）。 CreatedAt time.Time `json:"created_at"` // SourceKind 表示日志来源类型（例如 trap/syslog 等）。 diff --git a/internal/models/schema_version.go b/internal/models/schema_version.go index 9c91324..de144f4 100644 --- a/internal/models/schema_version.go +++ b/internal/models/schema_version.go @@ -6,14 +6,20 @@ import ( "gorm.io/gorm" ) -const requiredSchemaScript = "0001_baseline.up.sql" +type schemaMigration struct { + script string + checksum string +} var requiredSchema = struct { - checksum string - tables []string + migrations []schemaMigration + tables []string }{ - checksum: "092328c2cbd4e6316ae0e7fe2dd0f986ba30daf712945b8fa14be7f54c01f286", - tables: []string{"logs_events", "logs_syslog_rules", "logs_trap_rules", "logs_trace_spans"}, + migrations: []schemaMigration{ + {script: "0001_baseline.up.sql", checksum: "f6632f4927b205caf88bdbcf5f9db755ed4a174caacd4900d1aac12839fb9c5e"}, + {script: "0002_otlp_ingest_key.up.sql", checksum: "85923db07fea3e0dfaa33af0c2a3775b917c1463717c01300e0958602efeec1d"}, + }, + tables: []string{"logs_events", "logs_syslog_rules", "logs_trap_rules", "logs_trace_spans"}, } // RequireSchemaVersion 确认当前数据库已经由受控迁移脚本初始化。 @@ -22,19 +28,22 @@ func RequireSchemaVersion(db *gorm.DB, service string) error { return fmt.Errorf("%s 数据库连接未初始化", service) } - var count int64 - err := db.Raw(` + for _, migration := range requiredSchema.migrations { + var count int64 + err := db.Raw(` SELECT COUNT(*) FROM public.ops_schema_migrations WHERE service = ? AND script_name = ? AND checksum = ? AND status = 'success'`, - service, requiredSchemaScript, requiredSchema.checksum).Scan(&count).Error - if err != nil { - return fmt.Errorf("%s 数据库版本检查失败，请先执行 migrate-all.sh up --service %s: %w", service, service, err) - } - if count != 1 { - return fmt.Errorf("%s 数据库迁移版本不匹配，请执行 migrate-all.sh status --service %s", service, service) + service, migration.script, migration.checksum).Scan(&count).Error + if err != nil { + return fmt.Errorf("%s 数据库版本检查失败，请先执行 migrate-all.sh up --service %s: %w", service, service, err) + } + if count != 1 { + return fmt.Errorf("%s 数据库迁移版本不匹配：%s，请执行 migrate-all.sh status --service %s", service, migration.script, service) + } } + var err error for _, table := range requiredSchema.tables { var valid bool err = db.Raw(` @@ -49,5 +58,28 @@ SELECT to_regclass(?) IS NOT NULL AND EXISTS ( return fmt.Errorf("%s 数据库关键表结构不完整：%s", service, table) } } + + var ingestKeyValid bool + err = db.Raw(` +SELECT EXISTS ( + SELECT 1 + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'logs_events' + AND column_name = 'ingest_key' + AND data_type = 'character varying' + AND character_maximum_length = 160 + AND is_nullable = 'YES' +) AND EXISTS ( + SELECT 1 + FROM pg_indexes + WHERE schemaname = 'public' + AND tablename = 'logs_events' + AND indexname = 'idx_logs_events_ingest_key' + AND indexdef LIKE 'CREATE UNIQUE INDEX %' +)`).Scan(&ingestKeyValid).Error + if err != nil || !ingestKeyValid { + return fmt.Errorf("%s 数据库关键字段或索引不完整：logs_events.ingest_key", service) + } return nil } diff --git a/internal/models/trap_shield.go b/internal/models/trap_shield.go index d76fa91..32f9f64 100644 --- a/internal/models/trap_shield.go +++ b/internal/models/trap_shield.go @@ -1,29 +1,29 @@ -package models - -import "time" - -// TrapShield 表示一条针对 SNMP Trap 的“屏蔽/防护”规则。 -type TrapShield struct { - // ID 是数据库主键。 - ID uint `gorm:"primaryKey" json:"id"` - // CreatedAt 记录创建时间（GORM 自动维护）。 - CreatedAt time.Time `json:"created_at"` - // UpdatedAt 记录更新时间（GORM 自动维护）。 - UpdatedAt time.Time `json:"updated_at"` - // Name 规则名称，用于展示/标识。 - Name string `gorm:"size:256" json:"name"` - // Enabled 表示该规则是否启用。 - Enabled bool `gorm:"default:true" json:"enabled"` - // SourceIPCIDR 表示规则适用的源 IP 网段（CIDR）。 - SourceIPCIDR string `gorm:"size:64" json:"source_ip_cidr"` - // OIDPrefix 表示匹配的 OID 前缀。 - OIDPrefix string `gorm:"size:512" json:"oid_prefix"` - // InterfaceHint 关联提示信息（例如接口/线路标识），用于定位设备来源。 - InterfaceHint string `gorm:"size:256" json:"interface_hint"` - // TimeWindowsJSON 以 JSON 文本形式描述规则生效时间窗口。 - TimeWindowsJSON string `gorm:"type:text" json:"time_windows_json"` -} - -func (TrapShield) TableName() string { - return "logs_trap_shields" -} +package models + +import "time" + +// TrapShield 表示一条针对 SNMP Trap 的“屏蔽/防护”规则。 +type TrapShield struct { + // ID 是数据库主键。 + ID uint `gorm:"primaryKey" json:"id"` + // CreatedAt 记录创建时间（GORM 自动维护）。 + CreatedAt time.Time `json:"created_at"` + // UpdatedAt 记录更新时间（GORM 自动维护）。 + UpdatedAt time.Time `json:"updated_at"` + // Name 规则名称，用于展示/标识。 + Name string `gorm:"size:256;not null" json:"name"` + // Enabled 表示该规则是否启用。 + Enabled bool `gorm:"not null;default:true" json:"enabled"` + // SourceIPCIDR 表示规则适用的源 IP 网段（CIDR）。 + SourceIPCIDR string `gorm:"column:source_ip_cidr;size:64;not null;default:''" json:"source_ip_cidr"` + // OIDPrefixes 表示可匹配的 Trap OID 前缀；多个前缀之间为“或”关系。 + OIDPrefixes []string `gorm:"column:oid_prefixes;type:jsonb;serializer:json;not null;default:'[]'" json:"oid_prefixes"` + // InterfaceHint 关联提示信息（例如接口/线路标识），用于定位设备来源。 + InterfaceHint string `gorm:"size:256;not null;default:''" json:"interface_hint"` + // TimeWindowsJSON 以 JSON 文本形式描述规则生效时间窗口。 + TimeWindowsJSON string `gorm:"type:text;not null;default:''" json:"time_windows_json"` +} + +func (TrapShield) TableName() string { + return "logs_trap_shields" +} diff --git a/internal/routers/register.go b/internal/routers/register.go index e997146..f9a167f 100644 --- a/internal/routers/register.go +++ b/internal/routers/register.go @@ -45,6 +45,7 @@ func Register(srvKey string, engine *gin.Engine) { api.DELETE("/trap-dictionary/:id", controllers.DeleteTrapDictionary) api.GET("/trap-suppressions", controllers.ListTrapShields) + api.GET("/trap-suppression-options", controllers.ListTrapShieldOptions) api.POST("/trap-suppressions", controllers.CreateTrapShield) api.PUT("/trap-suppressions/:id", controllers.UpdateTrapShield) api.DELETE("/trap-suppressions/:id", controllers.DeleteTrapShield) @@ -53,6 +54,7 @@ func Register(srvKey string, engine *gin.Engine) { api.GET("/traces", otlp.ListTraces) api.GET("/traces/:trace_id", otlp.GetTrace) api.GET("/service-dependencies", otlp.ListDependencies) + api.GET("/middleware-candidates", otlp.ListMiddlewareCandidates) api.POST("/entries/:id/replay", controllers.ReplayLogEvent) api.GET("/alert-outbox", controllers.ListAlertOutbox) api.POST("/alert-outbox/:id/retry", controllers.RetryAlertOutbox) diff --git a/scripts/run_logs_e2e.py b/scripts/run_logs_e2e.py index d36e8f5..52c6c50 100644 --- a/scripts/run_logs_e2e.py +++ b/scripts/run_logs_e2e.py @@ -238,14 +238,14 @@ class Runner: "recovery_message": "recover", "enabled": True, } - suppression_body = { - "name": f"{suffix}-suppress", - "enabled": True, - "source_ip_cidr": "127.0.0.1/32", - "oid_prefix": "1.3.6.1.4.1.8072", - "interface_hint": "no-match", - "time_windows_json": "[]", - } + suppression_body = { + "name": f"{suffix}-suppress", + "enabled": True, + "source_ip_cidr": "127.0.0.1/32", + "oid_prefixes": ["1.3.6.1.4.1.8072"], + "interface_hint": "no-match", + "time_windows_json": "", + } created_ids: List[Tuple[str, int]] = [] try: s1, p1 = http_json("POST", f"{self.cfg.base_url}/syslog-rules", token=self.cfg.token, body=syslog_body) @@ -345,7 +345,7 @@ class Runner: for row in payload_obj(p0).get("items", []): if not row.get("enabled", False): continue - if str(row.get("source_ip_cidr", "")).strip() == "" and str(row.get("oid_prefix", "")).strip() == "" and str(row.get("interface_hint", "")).strip() == "" and str(row.get("time_windows_json", "")).strip() == "": + if str(row.get("source_ip_cidr", "")).strip() == "" and not row.get("oid_prefixes", []) and str(row.get("interface_hint", "")).strip() == "" and str(row.get("time_windows_json", "")).strip() == "": rid = int(row.get("id", 0)) if rid > 0: body = dict(row)