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

fix: 验收修正

Browse Source
This commit is contained in:
zxr
2026-09-21 10:46:34 +08:00
parent b0ffde3100
commit 972adb3d38
12 changed files with 506 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
internal/ingest/engine.go
View File

@@ -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()

34
internal/ingest/shield.go
View File

@@ -17,6 +17,38 @@ type timeWindow struct {
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 == "" {
@@ -35,7 +67,7 @@ func ipMatchesCIDR(ip net.IP, cidr string) bool {
func inTimeWindows(now time.Time, jsonStr string) bool {
s := strings.TrimSpace(jsonStr)
if s == "" || s == "null" {
if s == "" {
return true
}
var windows []timeWindow

8
internal/logic/controllers/crud.go
View File

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

53
internal/logic/controllers/rule_validation.go
View File

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

82
internal/logic/controllers/trap_shield_options.go Normal file
View File

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

9
internal/logic/otlp/ingest.go
View File

@@ -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",

112
internal/logic/otlp/query.go
View File

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

2
internal/models/log_event.go
View File

@@ -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 等）。

42
internal/models/schema_version.go
View File

@@ -6,13 +6,19 @@ import (
"gorm.io/gorm"
)
const requiredSchemaScript = "0001_baseline.up.sql"
type schemaMigration struct {
script string
checksum string
}
var requiredSchema = struct {
checksum string
migrations []schemaMigration
tables []string
}{
checksum: "092328c2cbd4e6316ae0e7fe2dd0f986ba30daf712945b8fa14be7f54c01f286",
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"},
}
@@ -22,19 +28,22 @@ func RequireSchemaVersion(db *gorm.DB, service string) error {
return fmt.Errorf("%s 数据库连接未初始化", service)
}
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
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 数据库迁移版本不匹配，请执行 migrate-all.sh status --service %s", service, service)
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
}

14
internal/models/trap_shield.go
View File

@@ -11,17 +11,17 @@ type TrapShield struct {
// UpdatedAt 记录更新时间（GORM 自动维护）。
UpdatedAt time.Time `json:"updated_at"`
// Name 规则名称，用于展示/标识。
Name string `gorm:"size:256" json:"name"`
Name string `gorm:"size:256;not null" json:"name"`
// Enabled 表示该规则是否启用。
Enabled bool `gorm:"default:true" json:"enabled"`
Enabled bool `gorm:"not null;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"`
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" json:"interface_hint"`
InterfaceHint string `gorm:"size:256;not null;default:''" json:"interface_hint"`
// TimeWindowsJSON 以 JSON 文本形式描述规则生效时间窗口。
TimeWindowsJSON string `gorm:"type:text" json:"time_windows_json"`
TimeWindowsJSON string `gorm:"type:text;not null;default:''" json:"time_windows_json"`
}
func (TrapShield) TableName() string {

2
internal/routers/register.go
View File

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

6
scripts/run_logs_e2e.py
View File

@@ -242,9 +242,9 @@ class Runner:
"name": f"{suffix}-suppress",
"enabled": True,
"source_ip_cidr": "127.0.0.1/32",
"oid_prefix": "1.3.6.1.4.1.8072",
"oid_prefixes": ["1.3.6.1.4.1.8072"],
"interface_hint": "no-match",
"time_windows_json": "[]",
"time_windows_json": "",
}
created_ids: List[Tuple[str, int]] = []
try:
@@ -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)