From b0ffde31001b47f181dc7413d4e6812d800f9258 Mon Sep 17 00:00:00 2001 From: zxr <271055687@qq.com> Date: Wed, 9 Sep 2026 16:45:01 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=88=9D=E9=AA=8C=E9=92=88=E5=AF=B9?= =?UTF-8?q?=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/main/main.go | 154 ++++++--- etc/logs_dev.yaml | 13 +- etc/logs_prod.yaml | 22 +- etc/ops-logs.service | 4 +- go.mod | 3 + internal/config/config.go | 104 +++++- internal/health/readiness.go | 115 +++++++ internal/impl/impl.go | 22 +- internal/ingest/alert_outbox.go | 32 +- internal/ingest/engine.go | 27 +- internal/ingest/lifecycle.go | 106 ++++++ internal/ingest/syslog_udp.go | 93 +++--- internal/ingest/trap_udp.go | 90 ++++-- internal/logic/audit/audit.go | 13 + internal/logic/audit/controller.go | 7 + internal/logic/controllers/crud.go | 14 + internal/logic/otlp/ingest.go | 502 +++++++++++++++++++++++++++++ internal/logic/otlp/query.go | 192 +++++++++++ internal/models/audit_log.go | 1 + internal/models/log_event.go | 5 + internal/models/query.go | 1 + internal/models/schema_version.go | 53 +++ internal/models/trace_span.go | 31 ++ internal/routers/register.go | 14 +- internal/systemsettings/sync.go | 71 ++++ 25 files changed, 1531 insertions(+), 158 deletions(-) create mode 100644 internal/health/readiness.go create mode 100644 internal/ingest/lifecycle.go create mode 100644 internal/logic/otlp/ingest.go create mode 100644 internal/logic/otlp/query.go create mode 100644 internal/models/schema_version.go create mode 100644 internal/models/trace_span.go create mode 100644 internal/systemsettings/sync.go diff --git a/cmd/main/main.go b/cmd/main/main.go index 5fb37be..5ad11d0 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -1,36 +1,118 @@ -package main - -import ( - "fmt" - - "git.apinb.com/bsm-sdk/core/infra" - "git.apinb.com/bsm-sdk/core/middleware" - "git.apinb.com/ops/logs/internal/config" - "git.apinb.com/ops/logs/internal/impl" - "git.apinb.com/ops/logs/internal/ingest" - "git.apinb.com/ops/logs/internal/routers" - "github.com/gin-gonic/gin" -) - -var ServiceKey = "Logs" - -func main() { - config.New(ServiceKey) - impl.NewImpl() - - ingest.StartRefresher() - ingest.StartAlertDispatcher() - ingest.StartSyslogUDP() - ingest.StartTrapUDP() - - app := gin.Default() - middleware.Mode(app) - app.Use(middleware.Cors()) - app.Use(gin.Recovery()) - app.HEAD("/", infra.Health) - routers.Register(ServiceKey, app) - - if err := app.Run(fmt.Sprintf(":%s", config.Spec.Port)); err != nil { - panic(err) - } -} +package main + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "git.apinb.com/bsm-sdk/core/infra" + "git.apinb.com/bsm-sdk/core/middleware" + "git.apinb.com/ops/logs/internal/config" + "git.apinb.com/ops/logs/internal/impl" + "git.apinb.com/ops/logs/internal/ingest" + "git.apinb.com/ops/logs/internal/models" + "git.apinb.com/ops/logs/internal/routers" + settingssync "git.apinb.com/ops/logs/internal/systemsettings" + "github.com/gin-gonic/gin" +) + +var ServiceKey = "Logs" + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + var backgroundJobs sync.WaitGroup + + config.New(ServiceKey) + settingsClient, err := settingssync.LoadInitial(ctx) + if err != nil { + panic(fmt.Errorf("logs: 初始化系统参数失败: %w", err)) + } + impl.NewImpl() + if err := models.RequireSchemaVersion(impl.DBService, "logs"); err != nil { + panic(err) + } + + if err := ingest.StartRefresher(ctx); err != nil { + shutdownAfterStartupFailure(stop, err) + } + ingest.StartAlertDispatcher(ctx) + if err := ingest.StartSyslogUDP(ctx); err != nil { + shutdownAfterStartupFailure(stop, err) + } + if err := ingest.StartTrapUDP(ctx); err != nil { + shutdownAfterStartupFailure(stop, err) + } + backgroundJobs.Add(1) + go func() { + defer backgroundJobs.Done() + settingssync.Start(ctx, settingsClient) + }() + + app := gin.Default() + middleware.Mode(app) + app.Use(middleware.Cors()) + app.Use(gin.Recovery()) + app.HEAD("/", infra.Health) + routers.Register(ServiceKey, app) + + server := &http.Server{Addr: config.Spec.Addr, Handler: app, ReadHeaderTimeout: 10 * time.Second} + errCh := make(chan error, 1) + go func() { + errCh <- server.ListenAndServe() + }() + + var serveErr error + select { + case err := <-errCh: + if err != nil && !errors.Is(err, http.ErrServerClosed) { + serveErr = err + } + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + serveErr = err + } + } + stop() + if !ingest.Wait(20 * time.Second) { + serveErr = errors.Join(serveErr, fmt.Errorf("后台 worker 未在退出超时内结束")) + } + if !waitForBackgroundJobs(&backgroundJobs, 20*time.Second) { + serveErr = errors.Join(serveErr, fmt.Errorf("系统参数任务未在退出超时内结束")) + } + if err := errors.Join(serveErr, impl.Close()); err != nil { + log.Printf("logs: 服务退出失败: %v", err) + panic(err) + } +} + +func waitForBackgroundJobs(wg *sync.WaitGroup, timeout time.Duration) bool { + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + return true + case <-time.After(timeout): + return false + } +} + +func shutdownAfterStartupFailure(stop context.CancelFunc, startErr error) { + stop() + if !ingest.Wait(20 * time.Second) { + startErr = errors.Join(startErr, fmt.Errorf("后台 worker 未在退出超时内结束")) + } + panic(errors.Join(startErr, impl.Close())) +} diff --git a/etc/logs_dev.yaml b/etc/logs_dev.yaml index e79adf3..c363e78 100644 --- a/etc/logs_dev.yaml +++ b/etc/logs_dev.yaml @@ -1,10 +1,11 @@ Service: logs -Port: 12440 +Port: 12440 +BindIP: 0.0.0.0 Databases: Driver: postgres Source: - - host=8.137.107.29 user=postgres password=Weidong2023~! dbname=ops_dev port=19432 sslmode=disable TimeZone=Asia/Shanghai + - "${OPS_LOGS_DB_DSN}" # - host=8.137.107.29 user=system password=12345678 dbname=ops port=54321 sslmode=disable TimeZone=Asia/Shanghai # cache DB的选择请在后面直接带参数，不带会自动HASH计算选择DB库。 @@ -29,3 +30,11 @@ AlertForward: ResourceEvent: hmac_secret: ${DC_CONTROL_LOGS_EVENT_SECRET} max_skew_secs: 300 + +OTLP: + internal_key: ${LOGS_OTLP_INTERNAL_KEY} + +SystemSettings: + base_url: http://localhost:12436 + refresh_seconds: 30 + timeout_seconds: 5 diff --git a/etc/logs_prod.yaml b/etc/logs_prod.yaml index e79adf3..7d257c3 100644 --- a/etc/logs_prod.yaml +++ b/etc/logs_prod.yaml @@ -1,14 +1,14 @@ Service: logs -Port: 12440 +Port: 12440 +BindIP: 127.0.0.1 Databases: - Driver: postgres - Source: - - host=8.137.107.29 user=postgres password=Weidong2023~! dbname=ops_dev port=19432 sslmode=disable TimeZone=Asia/Shanghai - # - host=8.137.107.29 user=system password=12345678 dbname=ops port=54321 sslmode=disable TimeZone=Asia/Shanghai + Driver: postgres + Source: + - "${OPS_LOGS_DB_DSN}" # cache DB的选择请在后面直接带参数，不带会自动HASH计算选择DB库。 -Cache: redis://null:Weidong2023~!@8.137.107.29:19379/ +Cache: ${OPS_CACHE_DSN} MicroService: Enable: false @@ -22,10 +22,18 @@ Ingest: AlertForward: enabled: true - base_url: https://ops-api.apinb.com + base_url: ${ALERT_BASE_URL} internal_key: ${LOGS_ALERT_SECRET} default_policy_id: 0 ResourceEvent: hmac_secret: ${DC_CONTROL_LOGS_EVENT_SECRET} max_skew_secs: 300 + +OTLP: + internal_key: ${LOGS_OTLP_INTERNAL_KEY} + +SystemSettings: + base_url: ${MGT_BASE_URL} + refresh_seconds: 30 + timeout_seconds: 5 diff --git a/etc/ops-logs.service b/etc/ops-logs.service index 57623e2..00f4c21 100644 --- a/etc/ops-logs.service +++ b/etc/ops-logs.service @@ -13,11 +13,11 @@ Environment=BSM_RuntimeMode=prod Environment=RUN_MODE=prod Environment=BSM_Prefix=/data/app ExecStart=/data/app/ops-logs -ExecStartPost=/data/app/systemd/wait-http.sh ops-logs http://127.0.0.1:12440/Logs/v1/ping/hello 60 +ExecStartPost=/data/app/systemd/wait-http.sh ops-logs http://127.0.0.1:12440/ready 60 Restart=on-failure RestartSec=5s TimeoutStartSec=75s -TimeoutStopSec=30s +TimeoutStopSec=75s KillSignal=SIGTERM StandardOutput=append:/data/app/logs/logs.log StandardError=inherit diff --git a/go.mod b/go.mod index 658df03..5e3d1b0 100644 --- a/go.mod +++ b/go.mod @@ -2,7 +2,10 @@ module git.apinb.com/ops/logs go 1.25.1 +replace git.apinb.com/ops/pkgs => ../pkgs + require ( + git.apinb.com/ops/pkgs v0.0.0 git.apinb.com/bsm-sdk/core v0.1.3 github.com/gin-gonic/gin v1.12.0 github.com/gosnmp/gosnmp v1.37.0 diff --git a/internal/config/config.go b/internal/config/config.go index 9f2bb46..baa83c0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,7 +1,10 @@ package config import ( + "fmt" "net" + "net/url" + "strconv" "strings" "git.apinb.com/bsm-sdk/core/conf" @@ -29,30 +32,101 @@ type ResourceEventConf struct { MaxSkewSecs int `yaml:"max_skew_secs"` } +type OTLPConf struct { + InternalKey string `yaml:"internal_key"` +} + +type SystemSettingsConf struct { + BaseURL string `yaml:"base_url"` + RefreshSeconds int `yaml:"refresh_seconds"` + TimeoutSeconds int `yaml:"timeout_seconds"` +} + type SrvConfig struct { - conf.Base `yaml:",inline"` - Databases *conf.DBConf `yaml:"Databases"` - MicroService *conf.MicroServiceConf `yaml:"MicroService"` - Rpc map[string]conf.RpcConf `yaml:"Rpc"` - Gateway *conf.GatewayConf `yaml:"Gateway"` - Apm *conf.ApmConf `yaml:"APM"` - Etcd *conf.EtcdConf `yaml:"Etcd"` - AlertForward *AlertForwardConf `yaml:"AlertForward"` - Ingest IngestConf `yaml:"Ingest"` - ResourceEvent ResourceEventConf `yaml:"ResourceEvent"` + conf.Base `yaml:",inline"` + Databases *conf.DBConf `yaml:"Databases"` + MicroService *conf.MicroServiceConf `yaml:"MicroService"` + Rpc map[string]conf.RpcConf `yaml:"Rpc"` + Gateway *conf.GatewayConf `yaml:"Gateway"` + Apm *conf.ApmConf `yaml:"APM"` + AlertForward *AlertForwardConf `yaml:"AlertForward"` + Ingest IngestConf `yaml:"Ingest"` + ResourceEvent ResourceEventConf `yaml:"ResourceEvent"` + OTLP OTLPConf `yaml:"OTLP"` + SystemSettings SystemSettingsConf `yaml:"SystemSettings"` } func New(srvKey string) { conf.New(srvKey, &Spec) - Spec.Port = conf.CheckPort(Spec.Port) - Spec.BindIP = conf.CheckIP(Spec.BindIP) + if err := Validate(); err != nil { + panic(fmt.Errorf("logs 配置校验失败: %w", err)) + } Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port) + conf.PrintInfo(Spec.Addr) +} + +// Validate 校验日志服务配置，不生成随机端口、监听地址或刷新间隔。 +func Validate() error { + Spec.Service = strings.TrimSpace(Spec.Service) + Spec.Port = strings.TrimSpace(Spec.Port) + Spec.BindIP = strings.TrimSpace(Spec.BindIP) Spec.ResourceEvent.HMACSecret = strings.TrimSpace(Spec.ResourceEvent.HMACSecret) - conf.NotNil(Spec.Service, Spec.Cache, Spec.ResourceEvent.HMACSecret) + Spec.OTLP.InternalKey = strings.TrimSpace(Spec.OTLP.InternalKey) + Spec.SystemSettings.BaseURL = strings.TrimRight(strings.TrimSpace(Spec.SystemSettings.BaseURL), "/") + if Spec.Service == "" { + return fmt.Errorf("Service 不能为空") + } + port, err := strconv.Atoi(Spec.Port) + if err != nil || port < 1 || port > 65535 { + return fmt.Errorf("Port 必须是 1 到 65535 的整数") + } + if net.ParseIP(Spec.BindIP) == nil { + return fmt.Errorf("BindIP 必须是明确的 IPv4 或 IPv6 地址") + } + if Spec.Databases == nil || strings.TrimSpace(Spec.Databases.Driver) == "" || len(Spec.Databases.Source) == 0 { + return fmt.Errorf("Databases.Driver 和 Databases.Source 不能为空") + } + if strings.TrimSpace(Spec.Cache) == "" { + return fmt.Errorf("Cache 不能为空") + } + if Spec.ResourceEvent.HMACSecret == "" || Spec.ResourceEvent.MaxSkewSecs <= 0 { + return fmt.Errorf("ResourceEvent.hmac_secret 和 max_skew_secs 必须有效") + } + if Spec.OTLP.InternalKey == "" { + return fmt.Errorf("OTLP.internal_key 不能为空") + } + parsedSystemSettingsURL, err := url.Parse(Spec.SystemSettings.BaseURL) + if err != nil || parsedSystemSettingsURL.Host == "" || (parsedSystemSettingsURL.Scheme != "http" && parsedSystemSettingsURL.Scheme != "https") { + return fmt.Errorf("SystemSettings.base_url 必须是完整的 HTTP 或 HTTPS 地址") + } + if Spec.SystemSettings.RefreshSeconds <= 0 || Spec.SystemSettings.TimeoutSeconds <= 0 { + return fmt.Errorf("SystemSettings.refresh_seconds 和 timeout_seconds 必须大于 0") + } + if Spec.Ingest.RuleRefreshSecs <= 0 { + return fmt.Errorf("Ingest.rule_refresh_secs 必须大于 0") + } + for name, addr := range map[string]string{ + "syslog_listen_addr": Spec.Ingest.SyslogListenAddr, + "trap_listen_addr": Spec.Ingest.TrapListenAddr, + } { + addr = strings.TrimSpace(addr) + if addr == "" { + continue + } + if _, err := net.ResolveUDPAddr("udp", addr); err != nil { + return fmt.Errorf("Ingest.%s 无效: %w", name, err) + } + } if Spec.AlertForward != nil && Spec.AlertForward.Enabled { Spec.AlertForward.BaseURL = strings.TrimSpace(Spec.AlertForward.BaseURL) Spec.AlertForward.InternalKey = strings.TrimSpace(Spec.AlertForward.InternalKey) - conf.NotNil(Spec.AlertForward.BaseURL, Spec.AlertForward.InternalKey) + parsed, err := url.Parse(Spec.AlertForward.BaseURL) + if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") { + return fmt.Errorf("AlertForward.base_url 必须是完整的 HTTP 或 HTTPS 地址") + } + if Spec.AlertForward.InternalKey == "" { + return fmt.Errorf("AlertForward.internal_key 不能为空") + } } - conf.PrintInfo(Spec.Addr) + return nil } diff --git a/internal/health/readiness.go b/internal/health/readiness.go new file mode 100644 index 0000000..0481ecd --- /dev/null +++ b/internal/health/readiness.go @@ -0,0 +1,115 @@ +package health + +import ( + "context" + "fmt" + "net/http" + "runtime/debug" + "strings" + "time" + + "git.apinb.com/ops/logs/internal/config" + "git.apinb.com/ops/logs/internal/impl" + "git.apinb.com/ops/logs/internal/ingest" + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +type CheckResult struct { + Name string `json:"name"` + Required bool `json:"required"` + Ready bool `json:"ready"` + Message string `json:"message,omitempty"` +} + +type Readiness struct { + Ready bool `json:"ready"` + CheckedAt time.Time `json:"checked_at"` + Checks []CheckResult `json:"checks,omitempty"` + Workers []ingest.WorkerStatus `json:"workers,omitempty"` + Version string `json:"version"` +} + +func Evaluate(ctx context.Context) Readiness { + result := Readiness{Ready: true, CheckedAt: time.Now().UTC(), Workers: ingest.WorkerStatuses(), Version: runtimeVersion()} + result.add("config", true, config.Validate()) + result.add("database", true, pingDatabase(ctx, impl.DBService)) + workerMap := make(map[string]ingest.WorkerStatus, len(result.Workers)) + for _, worker := range result.Workers { + workerMap[worker.Name] = worker + } + for _, name := range requiredWorkers() { + worker, exists := workerMap[name] + if !exists || !worker.Running { + message := "worker 未启动" + if exists && worker.LastError != "" { + message = worker.LastError + } + result.add(name, true, fmt.Errorf("%s", message)) + continue + } + result.add(name, true, nil) + } + return result +} + +func runtimeVersion() string { + info, ok := debug.ReadBuildInfo() + if !ok || strings.TrimSpace(info.Main.Version) == "" { + return "unknown" + } + return info.Main.Version +} + +func Ready(c *gin.Context) { + result := Evaluate(c.Request.Context()) + status := http.StatusOK + if !result.Ready { + status = http.StatusServiceUnavailable + } + c.JSON(status, gin.H{"ready": result.Ready}) +} + +func Status(c *gin.Context) { + result := Evaluate(c.Request.Context()) + status := http.StatusOK + if !result.Ready { + status = http.StatusServiceUnavailable + } + c.JSON(status, result) +} + +func requiredWorkers() []string { + workers := []string{"rule_refresher", "alert_dispatcher"} + if strings.TrimSpace(config.Spec.Ingest.SyslogListenAddr) != "" { + workers = append(workers, "syslog_udp") + } + if strings.TrimSpace(config.Spec.Ingest.TrapListenAddr) != "" { + workers = append(workers, "trap_udp") + } + return workers +} + +func (r *Readiness) add(name string, required bool, err error) { + check := CheckResult{Name: name, Required: required, Ready: err == nil} + if err != nil { + check.Message = strings.TrimSpace(err.Error()) + if required { + r.Ready = false + } + } + r.Checks = append(r.Checks, check) +} + +func pingDatabase(ctx context.Context, db *gorm.DB) error { + if db == nil { + return fmt.Errorf("数据库未初始化") + } + sqlDB, err := db.DB() + if err != nil { + return err + } + checkCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + return sqlDB.PingContext(checkCtx) +} diff --git a/internal/impl/impl.go b/internal/impl/impl.go index a3a3336..8b6c46a 100644 --- a/internal/impl/impl.go +++ b/internal/impl/impl.go @@ -1,12 +1,11 @@ package impl import ( - "fmt" + "errors" "git.apinb.com/bsm-sdk/core/cache/redis" "git.apinb.com/bsm-sdk/core/logger" "git.apinb.com/ops/logs/internal/config" - "git.apinb.com/ops/logs/internal/models" "gorm.io/gorm" ) @@ -19,13 +18,22 @@ func NewImpl() { RedisService = newRedisCache(config.Spec.Cache) DBService = newDatabase(config.Spec.Databases) logger.New(nil) +} +// Close 关闭数据库和 Redis 连接。 +func Close() error { + var closeErrors []error if DBService != nil { - if err := DBService.AutoMigrate(models.GetAllModels()...); err != nil { - panic(fmt.Sprintf("logs migrate: %v", err)) - } - if err := models.InitData(DBService); err != nil { - panic(fmt.Sprintf("logs init data: %v", err)) + if sqlDB, err := DBService.DB(); err != nil { + closeErrors = append(closeErrors, err) + } else if err := sqlDB.Close(); err != nil { + closeErrors = append(closeErrors, err) } } + if RedisService != nil && RedisService.Client != nil { + if err := RedisService.Client.Close(); err != nil { + closeErrors = append(closeErrors, err) + } + } + return errors.Join(closeErrors...) } diff --git a/internal/ingest/alert_outbox.go b/internal/ingest/alert_outbox.go index b33c435..028e46e 100644 --- a/internal/ingest/alert_outbox.go +++ b/internal/ingest/alert_outbox.go @@ -1,6 +1,7 @@ package ingest import ( + "context" "encoding/json" "errors" "fmt" @@ -63,17 +64,36 @@ func enqueuePayloadWithDB(db *gorm.DB, logEventID uint, payloadJSON string) (uin return row.ID, nil } -func StartAlertDispatcher() { +func StartAlertDispatcher(ctx context.Context) { owner := dispatcherOwner() - go func() { + startWorker("alert_dispatcher", func() error { ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() - for range ticker.C { - if _, err := ProcessAlertOutboxBatch(impl.DBService, 20, owner); err != nil { - log.Printf("logs: alert outbox dispatch: %v", err) + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + if _, err := ProcessAlertOutboxBatch(impl.DBService, 20, owner); err != nil { + markWorkerError("alert_dispatcher", err) + log.Printf("logs: alert outbox dispatch: %v", err) + continue + } + if depth, err := alertOutboxQueueDepth(ctx); err == nil { + markWorkerQueueDepth("alert_dispatcher", depth) + } + markWorkerSucceeded("alert_dispatcher") } } - }() + }) +} + +func alertOutboxQueueDepth(ctx context.Context) (int64, error) { + var depth int64 + err := impl.DBService.WithContext(ctx).Model(&models.AlertOutbox{}). + Where("status IN ?", []string{outboxStatusPending, outboxStatusRetrying, outboxStatusProcessing}). + Count(&depth).Error + return depth, err } func dispatcherOwner() string { diff --git a/internal/ingest/engine.go b/internal/ingest/engine.go index 24123be..61f0de1 100644 --- a/internal/ingest/engine.go +++ b/internal/ingest/engine.go @@ -1,6 +1,7 @@ package ingest import ( + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -138,19 +139,29 @@ func (e *Engine) Refresh() error { return nil } -func StartRefresher() { +func StartRefresher(ctx context.Context) error { interval := config.Spec.Ingest.RuleRefreshSecs - if interval <= 0 { - interval = 30 + if err := Global.Refresh(); err != nil { + return err } - _ = Global.Refresh() - go func() { + startWorker("rule_refresher", func() error { + markWorkerSucceeded("rule_refresher") t := time.NewTicker(time.Duration(interval) * time.Second) defer t.Stop() - for range t.C { - _ = Global.Refresh() + for { + select { + case <-ctx.Done(): + return nil + case <-t.C: + if err := Global.Refresh(); err != nil { + markWorkerError("rule_refresher", err) + continue + } + markWorkerSucceeded("rule_refresher") + } } - }() + }) + return nil } func normOID(s string) string { diff --git a/internal/ingest/lifecycle.go b/internal/ingest/lifecycle.go new file mode 100644 index 0000000..8efe695 --- /dev/null +++ b/internal/ingest/lifecycle.go @@ -0,0 +1,106 @@ +package ingest + +import ( + "sync" + "time" +) + +// WorkerStatus 描述日志后台 worker 的当前运行状态。 +type WorkerStatus struct { + Name string `json:"name"` + Running bool `json:"running"` + LastStartedAt time.Time `json:"last_started_at,omitempty"` + LastSucceededAt time.Time `json:"last_succeeded_at,omitempty"` + LastErrorAt time.Time `json:"last_error_at,omitempty"` + LastError string `json:"last_error,omitempty"` + QueueDepth *int64 `json:"queue_depth,omitempty"` +} + +func markWorkerQueueDepth(name string, depth int64) { + lifecycle.Lock() + status := lifecycle.workers[name] + status.Name = name + status.QueueDepth = new(int64) + *status.QueueDepth = depth + lifecycle.workers[name] = status + lifecycle.Unlock() +} + +var lifecycle = struct { + sync.RWMutex + wg sync.WaitGroup + workers map[string]WorkerStatus +}{workers: make(map[string]WorkerStatus)} + +func startWorker(name string, run func() error) { + lifecycle.Lock() + lifecycle.workers[name] = WorkerStatus{Name: name, Running: true, LastStartedAt: time.Now().UTC()} + lifecycle.wg.Add(1) + lifecycle.Unlock() + go func() { + defer lifecycle.wg.Done() + err := run() + lifecycle.Lock() + status := lifecycle.workers[name] + status.Running = false + if err != nil { + status.LastErrorAt = time.Now().UTC() + status.LastError = err.Error() + } + lifecycle.workers[name] = status + lifecycle.Unlock() + }() +} + +func markWorkerSucceeded(name string) { + lifecycle.Lock() + status := lifecycle.workers[name] + status.Name = name + status.LastSucceededAt = time.Now().UTC() + status.LastError = "" + lifecycle.workers[name] = status + lifecycle.Unlock() +} + +func markWorkerError(name string, err error) { + if err == nil { + return + } + lifecycle.Lock() + status := lifecycle.workers[name] + status.Name = name + status.LastErrorAt = time.Now().UTC() + status.LastError = err.Error() + lifecycle.workers[name] = status + lifecycle.Unlock() +} + +// WorkerStatuses 返回状态快照。 +func WorkerStatuses() []WorkerStatus { + lifecycle.RLock() + defer lifecycle.RUnlock() + result := make([]WorkerStatus, 0, len(lifecycle.workers)) + for _, status := range lifecycle.workers { + if status.QueueDepth != nil { + depth := *status.QueueDepth + status.QueueDepth = &depth + } + result = append(result, status) + } + return result +} + +// Wait 等待后台 worker 退出，返回是否在超时内完成。 +func Wait(timeout time.Duration) bool { + done := make(chan struct{}) + go func() { + lifecycle.wg.Wait() + close(done) + }() + select { + case <-done: + return true + case <-time.After(timeout): + return false + } +} diff --git a/internal/ingest/syslog_udp.go b/internal/ingest/syslog_udp.go index 139bca9..9c98a82 100644 --- a/internal/ingest/syslog_udp.go +++ b/internal/ingest/syslog_udp.go @@ -1,41 +1,52 @@ -package ingest - -import ( - "log" - "net" - "strings" - - "git.apinb.com/ops/logs/internal/config" -) - -func StartSyslogUDP() { - addr := strings.TrimSpace(config.Spec.Ingest.SyslogListenAddr) - if addr == "" { - return - } - go func() { - pc, err := net.ListenPacket("udp", addr) - if err != nil { - log.Printf("logs: syslog UDP listen %s: %v", addr, err) - return - } - defer pc.Close() - log.Printf("logs: syslog listening UDP %s", addr) - buf := make([]byte, 65536) - for { - n, remote, err := pc.ReadFrom(buf) - if err != nil { - log.Printf("logs: syslog read: %v", err) - continue - } - udpAddr, _ := remote.(*net.UDPAddr) - if udpAddr == nil { - continue - } - p := make([]byte, n) - copy(p, buf[:n]) - a := *udpAddr - Global.HandleSyslog(&a, p) - } - }() -} +package ingest + +import ( + "context" + "fmt" + "log" + "net" + "strings" + + "git.apinb.com/ops/logs/internal/config" +) + +func StartSyslogUDP(ctx context.Context) error { + addr := strings.TrimSpace(config.Spec.Ingest.SyslogListenAddr) + if addr == "" { + return nil + } + pc, err := net.ListenPacket("udp", addr) + if err != nil { + return fmt.Errorf("syslog UDP 监听 %s 失败: %w", addr, err) + } + startWorker("syslog_udp", func() error { + defer pc.Close() + go func() { + <-ctx.Done() + _ = pc.Close() + }() + log.Printf("logs: syslog listening UDP %s", addr) + buf := make([]byte, 65536) + for { + n, remote, err := pc.ReadFrom(buf) + if err != nil { + if ctx.Err() != nil { + return nil + } + markWorkerError("syslog_udp", err) + log.Printf("logs: syslog read: %v", err) + continue + } + udpAddr, _ := remote.(*net.UDPAddr) + if udpAddr == nil { + continue + } + p := make([]byte, n) + copy(p, buf[:n]) + a := *udpAddr + Global.HandleSyslog(&a, p) + markWorkerSucceeded("syslog_udp") + } + }) + return nil +} diff --git a/internal/ingest/trap_udp.go b/internal/ingest/trap_udp.go index 5a40364..40e6d48 100644 --- a/internal/ingest/trap_udp.go +++ b/internal/ingest/trap_udp.go @@ -1,32 +1,58 @@ -package ingest - -import ( - "log" - "net" - "strings" - - "git.apinb.com/ops/logs/internal/config" - "github.com/gosnmp/gosnmp" -) - -func StartTrapUDP() { - addr := strings.TrimSpace(config.Spec.Ingest.TrapListenAddr) - if addr == "" { - return - } - go func() { - tl := gosnmp.NewTrapListener() - tl.OnNewTrap = func(pkt *gosnmp.SnmpPacket, u *net.UDPAddr) { - if u == nil || pkt == nil { - return - } - ua := *u - Global.HandleTrap(&ua, pkt) - } - tl.Params = gosnmp.Default - tl.Params.Logger = gosnmp.NewLogger(log.Default()) - if err := tl.Listen(addr); err != nil { - log.Printf("logs: trap listener %s: %v", addr, err) - } - }() -} +package ingest + +import ( + "context" + "fmt" + "log" + "net" + "strings" + "time" + + "git.apinb.com/ops/logs/internal/config" + "github.com/gosnmp/gosnmp" +) + +func StartTrapUDP(ctx context.Context) error { + addr := strings.TrimSpace(config.Spec.Ingest.TrapListenAddr) + if addr == "" { + return nil + } + tl := gosnmp.NewTrapListener() + tl.OnNewTrap = func(pkt *gosnmp.SnmpPacket, u *net.UDPAddr) { + if u == nil || pkt == nil { + return + } + ua := *u + Global.HandleTrap(&ua, pkt) + markWorkerSucceeded("trap_udp") + } + tl.Params = gosnmp.Default + tl.Params.Logger = gosnmp.NewLogger(log.Default()) + started := make(chan error, 1) + startWorker("trap_udp", func() error { + go func() { + <-ctx.Done() + tl.Close() + }() + err := tl.Listen(addr) + if ctx.Err() != nil { + return nil + } + return err + }) + go func() { + select { + case <-tl.Listening(): + started <- nil + case <-ctx.Done(): + started <- ctx.Err() + case <-time.After(5 * time.Second): + tl.Close() + started <- fmt.Errorf("Trap UDP 监听 %s 启动超时", addr) + } + }() + if err := <-started; err != nil { + return err + } + return nil +} diff --git a/internal/logic/audit/audit.go b/internal/logic/audit/audit.go index fcfd916..2d76175 100644 --- a/internal/logic/audit/audit.go +++ b/internal/logic/audit/audit.go @@ -24,6 +24,7 @@ type Record struct { ActorID string `json:"actor_id,omitempty"` ActorName string `json:"actor_name,omitempty"` Action string `json:"action,omitempty"` + LogCategory string `json:"log_category,omitempty"` ObjectType string `json:"object_type,omitempty"` ObjectID string `json:"object_id,omitempty"` OperationRisk string `json:"operation_risk,omitempty"` @@ -61,12 +62,20 @@ func NormalizeRecord(record Record) Record { record.ActorID = strings.TrimSpace(record.ActorID) record.ActorName = strings.TrimSpace(record.ActorName) record.Action = strings.TrimSpace(record.Action) + record.LogCategory = strings.TrimSpace(strings.ToLower(record.LogCategory)) record.ObjectType = strings.TrimSpace(record.ObjectType) record.ObjectID = strings.TrimSpace(record.ObjectID) record.OperationRisk = strings.TrimSpace(strings.ToLower(record.OperationRisk)) record.ApprovalID = strings.TrimSpace(record.ApprovalID) record.RequestMethod = strings.TrimSpace(strings.ToUpper(record.RequestMethod)) record.RequestPath = strings.TrimSpace(record.RequestPath) + if record.LogCategory == "" { + if record.RequestMethod == "GET" || record.RequestMethod == "HEAD" { + record.LogCategory = "data" + } else { + record.LogCategory = "operation" + } + } record.ClientIP = strings.TrimSpace(record.ClientIP) record.Result = strings.TrimSpace(record.Result) if record.Result == "" { @@ -99,6 +108,9 @@ func ValidateRecord(record Record) error { if record.ObjectID == "" { return errors.New("object_id is required") } + if record.LogCategory != "operation" && record.LogCategory != "data" { + return errors.New("log_category must be operation or data") + } if record.OperationRisk != RiskNormal && record.OperationRisk != RiskDangerous { return errors.New("operation_risk must be normal or dangerous") } @@ -139,6 +151,7 @@ func SaveRecord(record Record) (models.AuditLog, error) { ActorID: record.ActorID, ActorName: record.ActorName, Action: record.Action, + LogCategory: record.LogCategory, ObjectType: record.ObjectType, ObjectID: record.ObjectID, OperationRisk: record.OperationRisk, diff --git a/internal/logic/audit/controller.go b/internal/logic/audit/controller.go index 2b9b423..b76fc5e 100644 --- a/internal/logic/audit/controller.go +++ b/internal/logic/audit/controller.go @@ -25,6 +25,13 @@ func ListAuditLogs(ctx *gin.Context) { if v := strings.TrimSpace(ctx.Query("action")); v != "" { q = q.Where("action = ?", v) } + if v := strings.TrimSpace(ctx.Query("log_category")); v != "" { + if v != "operation" && v != "data" { + infra.Response.Error(ctx, fmt.Errorf("log_category 参数无效")) + return + } + q = q.Where("log_category = ?", v) + } if v := strings.TrimSpace(ctx.Query("object_type")); v != "" { q = q.Where("object_type = ?", v) } diff --git a/internal/logic/controllers/crud.go b/internal/logic/controllers/crud.go index f6fcbcb..43ffb20 100644 --- a/internal/logic/controllers/crud.go +++ b/internal/logic/controllers/crud.go @@ -3,6 +3,7 @@ package controllers import ( "errors" "strconv" + "strings" "git.apinb.com/bsm-sdk/core/infra" "git.apinb.com/ops/logs/internal/impl" @@ -292,6 +293,9 @@ func ListLogEvents(ctx *gin.Context) { resourceType := ctx.Query("resource_type") resourceID := ctx.Query("resource_id") dispatchStatus := ctx.Query("dispatch_status") + traceID := strings.TrimSpace(ctx.Query("trace_id")) + resourceUID := strings.TrimSpace(ctx.Query("resource_uid")) + businessSystemID, _ := strconv.ParseUint(ctx.Query("business_system_id"), 10, 32) logEventID, _ := strconv.ParseUint(ctx.DefaultQuery("log_event_id", "0"), 10, 64) page, _ := strconv.Atoi(ctx.DefaultQuery("page", "1")) size, _ := strconv.Atoi(ctx.DefaultQuery("page_size", "50")) @@ -315,6 +319,16 @@ func ListLogEvents(ctx *gin.Context) { if dispatchStatus != "" { q = q.Where("dispatch_status = ?", dispatchStatus) } + if traceID != "" { + q = q.Where("trace_id = ?", strings.ToLower(traceID)) + } + if resourceUID != "" { + q = q.Where("resource_uid = ?", resourceUID) + } + if businessSystemID > 0 { + traceScope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("business_system_id = ?", uint(businessSystemID)) + q = q.Where("business_system_id = ? OR trace_id IN (?)", uint(businessSystemID), traceScope) + } if logEventID > 0 { q = q.Where("id = ?", uint(logEventID)) } diff --git a/internal/logic/otlp/ingest.go b/internal/logic/otlp/ingest.go new file mode 100644 index 0000000..ea49479 --- /dev/null +++ b/internal/logic/otlp/ingest.go @@ -0,0 +1,502 @@ +package otlp + +import ( + "crypto/subtle" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" + "unicode/utf8" + + "git.apinb.com/ops/logs/internal/config" + "git.apinb.com/ops/logs/internal/impl" + "git.apinb.com/ops/logs/internal/models" + "github.com/gin-gonic/gin" + "gorm.io/gorm/clause" +) + +const ( + maxRecordsPerRequest = 10000 + maxRequestBytes = 32 * 1024 * 1024 +) + +type anyValue struct { + StringValue *string `json:"stringValue"` + BoolValue *bool `json:"boolValue"` + IntValue json.RawMessage `json:"intValue"` + DoubleValue *float64 `json:"doubleValue"` + BytesValue string `json:"bytesValue"` + ArrayValue *arrayValue `json:"arrayValue"` + KVListValue *keyValueList `json:"kvlistValue"` +} + +type arrayValue struct { + Values []anyValue `json:"values"` +} + +type keyValueList struct { + Values []keyValue `json:"values"` +} + +type keyValue struct { + Key string `json:"key"` + Value anyValue `json:"value"` +} + +type resource struct { + Attributes []keyValue `json:"attributes"` +} + +type traceRequest struct { + ResourceSpans []struct { + Resource resource `json:"resource"` + ScopeSpans []struct { + Spans []spanPayload `json:"spans"` + } `json:"scopeSpans"` + } `json:"resourceSpans"` +} + +type spanPayload struct { + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` + ParentSpanID string `json:"parentSpanId"` + Name string `json:"name"` + Kind json.RawMessage `json:"kind"` + StartTimeUnixNano json.RawMessage `json:"startTimeUnixNano"` + EndTimeUnixNano json.RawMessage `json:"endTimeUnixNano"` + Attributes []keyValue `json:"attributes"` + Status struct { + Code json.RawMessage `json:"code"` + Message string `json:"message"` + } `json:"status"` +} + +type logRequest struct { + ResourceLogs []struct { + Resource resource `json:"resource"` + ScopeLogs []struct { + LogRecords []logPayload `json:"logRecords"` + } `json:"scopeLogs"` + } `json:"resourceLogs"` +} + +type logPayload struct { + TimeUnixNano json.RawMessage `json:"timeUnixNano"` + ObservedTimeUnixNano json.RawMessage `json:"observedTimeUnixNano"` + SeverityText string `json:"severityText"` + SeverityNumber json.RawMessage `json:"severityNumber"` + Body anyValue `json:"body"` + Attributes []keyValue `json:"attributes"` + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` +} + +func ReceiveTraces(ctx *gin.Context) { + if !authorize(ctx) { + return + } + ctx.Request.Body = http.MaxBytesReader(ctx.Writer, ctx.Request.Body, maxRequestBytes) + var request traceRequest + if err := ctx.ShouldBindJSON(&request); err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": "OTLP Trace 请求无效"}) + return + } + rows, rejected, message := decodeSpans(request) + if len(rows) > 0 { + if err := impl.DBService.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "trace_id"}, {Name: "span_id"}}, DoNothing: true}). + CreateInBatches(rows, 500).Error; err != nil { + ctx.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{"error": "Trace 入库失败"}) + return + } + } + ctx.JSON(http.StatusOK, gin.H{"partialSuccess": gin.H{"rejectedSpans": rejected, "errorMessage": message}}) +} + +func ReceiveLogs(ctx *gin.Context) { + if !authorize(ctx) { + return + } + ctx.Request.Body = http.MaxBytesReader(ctx.Writer, ctx.Request.Body, maxRequestBytes) + var request logRequest + if err := ctx.ShouldBindJSON(&request); err != nil { + ctx.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": "OTLP Log 请求无效"}) + return + } + rows, rejected, message := decodeLogs(request) + if len(rows) > 0 { + if err := impl.DBService.CreateInBatches(rows, 500).Error; err != nil { + ctx.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{"error": "Log 入库失败"}) + return + } + } + ctx.JSON(http.StatusOK, gin.H{"partialSuccess": gin.H{"rejectedLogRecords": rejected, "errorMessage": message}}) +} + +func authorize(ctx *gin.Context) bool { + provided := strings.TrimSpace(ctx.GetHeader("X-OTLP-Internal-Key")) + expected := config.Spec.OTLP.InternalKey + if provided == "" || len(provided) != len(expected) || subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) != 1 { + ctx.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "OTLP 接入密钥无效"}) + return false + } + return true +} + +func decodeSpans(request traceRequest) ([]models.TraceSpan, int, string) { + rows := make([]models.TraceSpan, 0) + rejected := 0 + for _, group := range request.ResourceSpans { + resourceAttrs := attributesMap(group.Resource.Attributes) + resourceJSON, _ := json.Marshal(resourceAttrs) + for _, scope := range group.ScopeSpans { + for _, input := range scope.Spans { + if len(rows)+rejected >= maxRecordsPerRequest { + rejected++ + continue + } + row, err := spanModel(input, resourceAttrs, string(resourceJSON)) + if err != nil { + rejected++ + continue + } + rows = append(rows, row) + } + } + } + message := "" + if rejected > 0 { + message = "部分 Span 因标识或时间字段无效被拒绝" + } + return rows, rejected, message +} + +func spanModel(input spanPayload, resourceAttrs map[string]interface{}, resourceJSON string) (models.TraceSpan, error) { + traceID, err := normalizeID(input.TraceID, 16) + if err != nil { + return models.TraceSpan{}, err + } + spanID, err := normalizeID(input.SpanID, 8) + if err != nil { + return models.TraceSpan{}, err + } + parentID := "" + if strings.TrimSpace(input.ParentSpanID) != "" { + parentID, err = normalizeID(input.ParentSpanID, 8) + if err != nil { + return models.TraceSpan{}, err + } + } + start, err := nanoTime(input.StartTimeUnixNano) + if err != nil { + return models.TraceSpan{}, err + } + end, err := nanoTime(input.EndTimeUnixNano) + if err != nil || end.Before(start) { + return models.TraceSpan{}, fmt.Errorf("Span 时间无效") + } + attrs := attributesMap(input.Attributes) + attrsJSON, _ := json.Marshal(attrs) + serviceName := textAttribute(resourceAttrs, "service.name") + if serviceName == "" { + serviceName = "unknown_service" + } + serviceName = limitCharacters(serviceName, 255) + operation := strings.TrimSpace(input.Name) + if operation == "" { + operation = "unnamed" + } + operation = limitCharacters(operation, 512) + statusCode := statusText(input.Status.Code) + if httpStatus := intAttribute(attrs, "http.response.status_code", "http.status_code"); statusCode == "unset" && httpStatus >= 500 { + statusCode = "error" + } + now := time.Now().UTC() + return models.TraceSpan{ + TraceID: traceID, SpanID: spanID, ParentSpanID: parentID, ServiceName: serviceName, + OperationName: operation, SpanKind: spanKindText(input.Kind), StartTime: start, EndTime: end, + DurationMs: float64(end.Sub(start).Nanoseconds()) / 1e6, StatusCode: statusCode, + StatusMessage: strings.TrimSpace(input.Status.Message), BusinessSystemID: uintAttribute(resourceAttrs, attrs, "business.system.id", "business_system_id"), + ResourceUID: limitCharacters(firstTextAttribute(resourceAttrs, attrs, "resource.uid", "resource_uid"), 255), + HTTPMethod: limitCharacters(firstTextAttribute(attrs, nil, "http.request.method", "http.method"), 16), + HTTPRoute: limitCharacters(firstTextAttribute(attrs, nil, "http.route", "url.path", "http.target"), 1024), + HTTPStatusCode: intAttribute(attrs, "http.response.status_code", "http.status_code"), + Attributes: string(attrsJSON), ResourceAttrs: resourceJSON, IngestedAt: &now, + }, nil +} + +func decodeLogs(request logRequest) ([]models.LogEvent, int, string) { + rows := make([]models.LogEvent, 0) + rejected := 0 + for _, group := range request.ResourceLogs { + resourceAttrs := attributesMap(group.Resource.Attributes) + for _, scope := range group.ScopeLogs { + for _, input := range scope.LogRecords { + if len(rows)+rejected >= maxRecordsPerRequest { + rejected++ + continue + } + row, err := logModel(input, resourceAttrs) + if err != nil { + rejected++ + continue + } + rows = append(rows, row) + } + } + } + message := "" + if rejected > 0 { + message = "部分 LogRecord 因标识或内容无效被拒绝" + } + return rows, rejected, message +} + +func logModel(input logPayload, resourceAttrs map[string]interface{}) (models.LogEvent, error) { + attrs := attributesMap(input.Attributes) + detailJSON, _ := json.Marshal(attrs) + body := scalarText(anyValueToInterface(input.Body)) + if body == "" { + return models.LogEvent{}, fmt.Errorf("日志正文为空") + } + if len(body) > 64*1024 { + body = limitUTF8Bytes(body, 64*1024) + } + traceID := "" + spanID := "" + var err error + if strings.TrimSpace(input.TraceID) != "" { + traceID, err = normalizeID(input.TraceID, 16) + if err != nil { + return models.LogEvent{}, err + } + } + if strings.TrimSpace(input.SpanID) != "" { + spanID, err = normalizeID(input.SpanID, 8) + if err != nil { + return models.LogEvent{}, err + } + } + serviceName := limitCharacters(textAttribute(resourceAttrs, "service.name"), 512) + resourceUID := limitCharacters(firstTextAttribute(resourceAttrs, attrs, "resource.uid", "resource_uid"), 255) + return models.LogEvent{ + 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", + SeverityCode: limitCharacters(severityText(input.SeverityText, input.SeverityNumber), 32), + }, nil +} + +func limitCharacters(value string, maximum int) string { + characters := []rune(value) + if len(characters) <= maximum { + return value + } + return string(characters[:maximum]) +} + +func limitUTF8Bytes(value string, maximum int) string { + if len(value) <= maximum { + return value + } + end := maximum + for end > 0 && !utf8.ValidString(value[:end]) { + end-- + } + return value[:end] +} + +func attributesMap(values []keyValue) map[string]interface{} { + result := make(map[string]interface{}, len(values)) + for _, item := range values { + if key := strings.TrimSpace(item.Key); key != "" { + result[key] = anyValueToInterface(item.Value) + } + } + return result +} + +func anyValueToInterface(value anyValue) interface{} { + if value.StringValue != nil { + return *value.StringValue + } + if value.BoolValue != nil { + return *value.BoolValue + } + if value.DoubleValue != nil { + return *value.DoubleValue + } + if len(value.IntValue) > 0 { + var number json.Number + if json.Unmarshal(value.IntValue, &number) == nil { + return number + } + var text string + if json.Unmarshal(value.IntValue, &text) == nil { + return text + } + } + if value.BytesValue != "" { + return value.BytesValue + } + if value.ArrayValue != nil { + items := make([]interface{}, 0, len(value.ArrayValue.Values)) + for _, item := range value.ArrayValue.Values { + items = append(items, anyValueToInterface(item)) + } + return items + } + if value.KVListValue != nil { + return attributesMap(value.KVListValue.Values) + } + return nil +} + +func normalizeID(value string, size int) (string, error) { + value = strings.TrimSpace(value) + if raw, err := hex.DecodeString(value); err == nil && len(raw) == size { + return strings.ToLower(value), nil + } + raw, err := base64.StdEncoding.DecodeString(value) + if err != nil || len(raw) != size { + return "", fmt.Errorf("OTLP 标识无效") + } + return hex.EncodeToString(raw), nil +} + +func nanoTime(raw json.RawMessage) (time.Time, error) { + value := strings.Trim(strings.TrimSpace(string(raw)), `"`) + nanos, err := strconv.ParseInt(value, 10, 64) + if err != nil || nanos <= 0 { + return time.Time{}, fmt.Errorf("OTLP 时间无效") + } + return time.Unix(0, nanos).UTC(), nil +} + +func enumValue(raw json.RawMessage) string { + return strings.Trim(strings.TrimSpace(string(raw)), `"`) +} + +func spanKindText(raw json.RawMessage) string { + switch strings.ToUpper(enumValue(raw)) { + case "2", "SPAN_KIND_SERVER": + return "server" + case "3", "SPAN_KIND_CLIENT": + return "client" + case "4", "SPAN_KIND_PRODUCER": + return "producer" + case "5", "SPAN_KIND_CONSUMER": + return "consumer" + default: + return "internal" + } +} + +func statusText(raw json.RawMessage) string { + switch strings.ToUpper(enumValue(raw)) { + case "1", "STATUS_CODE_OK": + return "ok" + case "2", "STATUS_CODE_ERROR": + return "error" + default: + return "unset" + } +} + +func severityText(text string, raw json.RawMessage) string { + text = strings.ToLower(strings.TrimSpace(text)) + if text != "" { + if strings.Contains(text, "fatal") { + return "critical" + } + if strings.Contains(text, "error") { + return "major" + } + if strings.Contains(text, "warn") { + return "warning" + } + return "info" + } + value, _ := strconv.Atoi(enumValue(raw)) + if value >= 21 { + return "critical" + } + if value >= 17 { + return "major" + } + if value >= 13 { + return "warning" + } + return "info" +} + +func timeFromLog(input logPayload) time.Time { + if value, err := nanoTime(input.TimeUnixNano); err == nil { + return value + } + if value, err := nanoTime(input.ObservedTimeUnixNano); err == nil { + return value + } + return time.Now().UTC() +} + +func textAttribute(values map[string]interface{}, key string) string { + return scalarText(values[key]) +} + +func firstTextAttribute(primary, secondary map[string]interface{}, keys ...string) string { + for _, key := range keys { + if primary != nil { + if value := scalarText(primary[key]); value != "" { + return value + } + } + if secondary != nil { + if value := scalarText(secondary[key]); value != "" { + return value + } + } + } + return "" +} + +func intAttribute(values map[string]interface{}, keys ...string) int { + for _, key := range keys { + value, err := strconv.Atoi(scalarText(values[key])) + if err == nil && value != 0 { + return value + } + } + return 0 +} + +func uintAttribute(primary, secondary map[string]interface{}, keys ...string) *uint { + value := firstTextAttribute(primary, secondary, keys...) + parsed, err := strconv.ParseUint(value, 10, 32) + if err != nil || parsed == 0 { + return nil + } + result := uint(parsed) + return &result +} + +func scalarText(value interface{}) string { + switch typed := value.(type) { + case string: + return typed + case json.Number: + return typed.String() + case float64, bool: + return fmt.Sprint(typed) + default: + if typed == nil { + return "" + } + encoded, _ := json.Marshal(typed) + return string(encoded) + } +} diff --git a/internal/logic/otlp/query.go b/internal/logic/otlp/query.go new file mode 100644 index 0000000..6f27159 --- /dev/null +++ b/internal/logic/otlp/query.go @@ -0,0 +1,192 @@ +package otlp + +import ( + "encoding/json" + "errors" + "sort" + "strconv" + "strings" + "time" + + "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" + "gorm.io/gorm" +) + +type traceSummary struct { + TraceID string `json:"trace_id"` + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + DurationMs float64 `json:"duration_ms"` + SpanCount int64 `json:"span_count"` + ErrorCount int64 `json:"error_count"` + Services string `json:"services"` + BusinessSystemID *uint `json:"business_system_id,omitempty"` + ResourceUID string `json:"resource_uid,omitempty"` +} + +type dependencyEdge struct { + SourceService string `json:"source_service"` + TargetService string `json:"target_service"` + CallCount int64 `json:"call_count"` + ErrorCount int64 `json:"error_count"` + AverageMs float64 `json:"average_ms"` +} + +type dependencyAggregate struct { + Calls int64 + Errors int64 + DurationMs float64 +} + +func ListTraces(ctx *gin.Context) { + page, size := pageParams(ctx) + query := applyTraceFilters(impl.DBService.Model(&models.TraceSpan{}), ctx) + var total int64 + if err := query.Distinct("trace_id").Count(&total).Error; err != nil { + infra.Response.Error(ctx, err) + return + } + var rows []traceSummary + err := query.Select(`trace_id, MIN(start_time) AS start_time, MAX(end_time) AS end_time, +EXTRACT(EPOCH FROM (MAX(end_time) - MIN(start_time))) * 1000 AS duration_ms, COUNT(*) AS span_count, +COUNT(*) FILTER (WHERE status_code = 'error') AS error_count, +STRING_AGG(DISTINCT service_name, ', ' ORDER BY service_name) AS services, +MAX(business_system_id) AS business_system_id, MAX(resource_uid) AS resource_uid`). + Group("trace_id").Order("start_time DESC").Offset((page - 1) * size).Limit(size).Scan(&rows).Error + if err != nil { + infra.Response.Error(ctx, err) + return + } + infra.Response.Success(ctx, gin.H{"total": total, "page": page, "page_size": size, "items": rows}) +} + +func GetTrace(ctx *gin.Context) { + traceID := strings.ToLower(strings.TrimSpace(ctx.Param("trace_id"))) + if _, err := normalizeID(traceID, 16); err != nil { + infra.Response.Error(ctx, errors.New("trace_id 无效")) + return + } + var rows []models.TraceSpan + if err := impl.DBService.Where("trace_id = ?", traceID).Order("start_time ASC, id ASC").Find(&rows).Error; err != nil { + infra.Response.Error(ctx, err) + return + } + if len(rows) == 0 { + infra.Response.Error(ctx, gorm.ErrRecordNotFound) + return + } + infra.Response.Success(ctx, gin.H{"trace_id": traceID, "spans": rows, "count": len(rows)}) +} + +func ListDependencies(ctx *gin.Context) { + query := applyTraceFilters(impl.DBService.Model(&models.TraceSpan{}), ctx) + var rows []models.TraceSpan + if err := query.Order("start_time DESC").Limit(100000).Find(&rows).Error; err != nil { + infra.Response.Error(ctx, err) + return + } + spanIndex := make(map[string]models.TraceSpan, len(rows)) + for _, row := range rows { + spanIndex[row.TraceID+":"+row.SpanID] = row + } + aggregates := make(map[string]*dependencyAggregate) + for _, row := range rows { + target := "" + if row.ParentSpanID != "" { + if parent, exists := spanIndex[row.TraceID+":"+row.ParentSpanID]; exists && parent.ServiceName != row.ServiceName { + target = row.ServiceName + addDependency(aggregates, parent.ServiceName, target, row) + continue + } + } + if row.SpanKind == "client" { + var attrs map[string]interface{} + _ = json.Unmarshal([]byte(row.Attributes), &attrs) + target = firstTextAttribute(attrs, nil, "peer.service", "server.address", "network.peer.address") + if target != "" && target != row.ServiceName { + addDependency(aggregates, row.ServiceName, target, row) + } + } + } + edges := make([]dependencyEdge, 0, len(aggregates)) + for key, value := range aggregates { + parts := strings.SplitN(key, "\x00", 2) + edges = append(edges, dependencyEdge{ + SourceService: parts[0], TargetService: parts[1], CallCount: value.Calls, + ErrorCount: value.Errors, AverageMs: value.DurationMs / float64(value.Calls), + }) + } + sort.Slice(edges, func(left, right int) bool { + if edges[left].SourceService != edges[right].SourceService { + return edges[left].SourceService < edges[right].SourceService + } + return edges[left].TargetService < edges[right].TargetService + }) + infra.Response.Success(ctx, gin.H{"items": edges, "count": len(edges)}) +} + +func addDependency(aggregates map[string]*dependencyAggregate, source, target string, span models.TraceSpan) { + key := source + "\x00" + target + item := aggregates[key] + if item == nil { + item = &dependencyAggregate{} + aggregates[key] = item + } + item.Calls++ + item.DurationMs += span.DurationMs + if span.StatusCode == "error" { + item.Errors++ + } +} + +func applyTraceFilters(query *gorm.DB, ctx *gin.Context) *gorm.DB { + if value := strings.TrimSpace(ctx.Query("trace_id")); value != "" { + query = query.Where("trace_id = ?", strings.ToLower(value)) + } + if value := strings.TrimSpace(ctx.Query("service_name")); value != "" { + scope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("service_name ILIKE ?", "%"+value+"%") + query = query.Where("trace_id IN (?)", scope) + } + if value := strings.TrimSpace(ctx.Query("resource_uid")); value != "" { + scope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("resource_uid = ?", value) + query = query.Where("trace_id IN (?)", scope) + } + if value, err := strconv.ParseUint(ctx.Query("business_system_id"), 10, 32); err == nil && value > 0 { + scope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("business_system_id = ?", uint(value)) + query = query.Where("trace_id IN (?)", scope) + } + if value := strings.ToLower(strings.TrimSpace(ctx.Query("status"))); value == "error" { + errorScope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("status_code = ?", "error") + query = query.Where("trace_id IN (?)", errorScope) + } else if value == "ok" { + errorScope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("status_code = ?", "error") + query = query.Where("trace_id NOT IN (?)", errorScope) + } else if value == "unset" { + setScope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("status_code IN ?", []string{"ok", "error"}) + query = query.Where("trace_id NOT IN (?)", setScope) + } + start := time.Now().UTC().Add(-24 * time.Hour) + if value, err := time.Parse(time.RFC3339, ctx.Query("start_time")); err == nil { + start = value.UTC() + } + timeScope := impl.DBService.Model(&models.TraceSpan{}).Select("trace_id").Where("start_time >= ?", start) + if value, err := time.Parse(time.RFC3339, ctx.Query("end_time")); err == nil { + timeScope = timeScope.Where("start_time <= ?", value.UTC()) + } + return query.Where("trace_id IN (?)", timeScope) +} + +func pageParams(ctx *gin.Context) (int, int) { + page, _ := strconv.Atoi(ctx.DefaultQuery("page", "1")) + size, _ := strconv.Atoi(ctx.DefaultQuery("page_size", "50")) + if page < 1 { + page = 1 + } + if size < 1 || size > 500 { + size = 50 + } + return page, size +} diff --git a/internal/models/audit_log.go b/internal/models/audit_log.go index fbccd7e..6890486 100644 --- a/internal/models/audit_log.go +++ b/internal/models/audit_log.go @@ -10,6 +10,7 @@ type AuditLog struct { ActorID string `gorm:"size:128;index" json:"actor_id"` ActorName string `gorm:"size:128" json:"actor_name"` Action string `gorm:"size:128;index" json:"action"` + LogCategory string `gorm:"size:16;index" json:"log_category"` ObjectType string `gorm:"size:128;index" json:"object_type"` ObjectID string `gorm:"size:128;index" json:"object_id"` OperationRisk string `gorm:"size:32;index" json:"operation_risk"` diff --git a/internal/models/log_event.go b/internal/models/log_event.go index 978f7c2..a3ae056 100644 --- a/internal/models/log_event.go +++ b/internal/models/log_event.go @@ -30,6 +30,11 @@ type LogEvent struct { ResourceID string `gorm:"size:128;index" json:"resource_id"` // ResourceName 表示关联到的资源名称。 ResourceName string `gorm:"size:256" json:"resource_name"` + // BusinessSystemID 关联业务系统。 + BusinessSystemID *uint `gorm:"index" json:"business_system_id,omitempty"` + // TraceID 和 SpanID 用于从日志定位调用链。 + TraceID string `gorm:"size:32;index" json:"trace_id,omitempty"` + SpanID string `gorm:"size:16;index" json:"span_id,omitempty"` // MatchMethod 表示资源命中方式（ip/hostname/none）。 MatchMethod string `gorm:"size:32" json:"match_method"` // DispatchStatus 表示告警分发状态（not_applicable/pending/retrying/sent/dead）。 diff --git a/internal/models/query.go b/internal/models/query.go index 58e6901..95279e9 100644 --- a/internal/models/query.go +++ b/internal/models/query.go @@ -10,6 +10,7 @@ import ( func GetAllModels() []interface{} { return []interface{}{ &LogEvent{}, + &TraceSpan{}, &AlertOutbox{}, &ResourceMapping{}, &ResourceEventDedup{}, diff --git a/internal/models/schema_version.go b/internal/models/schema_version.go new file mode 100644 index 0000000..9c91324 --- /dev/null +++ b/internal/models/schema_version.go @@ -0,0 +1,53 @@ +package models + +import ( + "fmt" + + "gorm.io/gorm" +) + +const requiredSchemaScript = "0001_baseline.up.sql" + +var requiredSchema = struct { + checksum string + tables []string +}{ + checksum: "092328c2cbd4e6316ae0e7fe2dd0f986ba30daf712945b8fa14be7f54c01f286", + tables: []string{"logs_events", "logs_syslog_rules", "logs_trap_rules", "logs_trace_spans"}, +} + +// RequireSchemaVersion 确认当前数据库已经由受控迁移脚本初始化。 +func RequireSchemaVersion(db *gorm.DB, service string) error { + if db == nil { + return fmt.Errorf("%s 数据库连接未初始化", service) + } + + 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) + } + + for _, table := range requiredSchema.tables { + var valid bool + err = db.Raw(` +SELECT to_regclass(?) IS NOT NULL AND EXISTS ( + SELECT 1 + FROM pg_constraint c + JOIN pg_class t ON t.oid = c.conrelid + JOIN pg_namespace n ON n.oid = t.relnamespace + WHERE n.nspname = 'public' AND t.relname = ? AND c.contype = 'p' +)`, "public."+table, table).Scan(&valid).Error + if err != nil || !valid { + return fmt.Errorf("%s 数据库关键表结构不完整：%s", service, table) + } + } + return nil +} diff --git a/internal/models/trace_span.go b/internal/models/trace_span.go new file mode 100644 index 0000000..51bbf3d --- /dev/null +++ b/internal/models/trace_span.go @@ -0,0 +1,31 @@ +package models + +import "time" + +// TraceSpan 保存通过 OTLP 接收的标准调用链 Span。 +type TraceSpan struct { + ID uint `gorm:"primaryKey" json:"id"` + CreatedAt time.Time `json:"created_at"` + + TraceID string `gorm:"type:varchar(32);not null;index;uniqueIndex:uk_trace_span" json:"trace_id"` + SpanID string `gorm:"type:varchar(16);not null;uniqueIndex:uk_trace_span" json:"span_id"` + ParentSpanID string `gorm:"type:varchar(16);index" json:"parent_span_id,omitempty"` + ServiceName string `gorm:"type:varchar(255);not null;index" json:"service_name"` + OperationName string `gorm:"type:varchar(512);not null;index" json:"operation_name"` + SpanKind string `gorm:"type:varchar(20);not null;index" json:"span_kind"` + StartTime time.Time `gorm:"not null;index" json:"start_time"` + EndTime time.Time `gorm:"not null;index" json:"end_time"` + DurationMs float64 `gorm:"type:decimal(16,3);not null;default:0" json:"duration_ms"` + StatusCode string `gorm:"type:varchar(20);not null;index" json:"status_code"` + StatusMessage string `gorm:"type:text" json:"status_message,omitempty"` + BusinessSystemID *uint `gorm:"index" json:"business_system_id,omitempty"` + ResourceUID string `gorm:"type:varchar(255);index" json:"resource_uid,omitempty"` + HTTPMethod string `gorm:"type:varchar(16);index" json:"http_method,omitempty"` + HTTPRoute string `gorm:"type:varchar(1024)" json:"http_route,omitempty"` + HTTPStatusCode int `gorm:"index" json:"http_status_code,omitempty"` + Attributes string `gorm:"type:jsonb;not null;default:'{}'" json:"attributes"` + ResourceAttrs string `gorm:"type:jsonb;not null;default:'{}'" json:"resource_attributes"` + IngestedAt *time.Time `gorm:"index" json:"ingested_at,omitempty"` +} + +func (TraceSpan) TableName() string { return "logs_trace_spans" } diff --git a/internal/routers/register.go b/internal/routers/register.go index 8228d3a..e997146 100644 --- a/internal/routers/register.go +++ b/internal/routers/register.go @@ -4,18 +4,27 @@ import ( "fmt" "git.apinb.com/bsm-sdk/core/middleware" + "git.apinb.com/ops/logs/internal/health" "git.apinb.com/ops/logs/internal/logic/audit" "git.apinb.com/ops/logs/internal/logic/controllers" + "git.apinb.com/ops/logs/internal/logic/otlp" "git.apinb.com/ops/logs/internal/logic/ping" "github.com/gin-gonic/gin" ) func Register(srvKey string, engine *gin.Engine) { v1 := fmt.Sprintf("/%s/%s", srvKey, "v1") + engine.GET("/ready", health.Ready) anon := engine.Group(v1) { anon.GET("/ping/hello", ping.Hello) + anon.POST("/resource-events", controllers.ReceiveResourceEvent) + anon.POST("/otlp/v1/traces", otlp.ReceiveTraces) + anon.POST("/otlp/v1/logs", otlp.ReceiveLogs) } + runtimeStatus := engine.Group(v1) + runtimeStatus.Use(middleware.JwtAuth(true)) + runtimeStatus.GET("/runtime/status", health.Status) api := engine.Group(v1) api.Use(middleware.JwtAuth(true)) @@ -40,9 +49,10 @@ func Register(srvKey string, engine *gin.Engine) { api.PUT("/trap-suppressions/:id", controllers.UpdateTrapShield) api.DELETE("/trap-suppressions/:id", controllers.DeleteTrapShield) - api.POST("/resource-events", controllers.ReceiveResourceEvent) - api.GET("/entries", controllers.ListLogEvents) + api.GET("/traces", otlp.ListTraces) + api.GET("/traces/:trace_id", otlp.GetTrace) + api.GET("/service-dependencies", otlp.ListDependencies) api.POST("/entries/:id/replay", controllers.ReplayLogEvent) api.GET("/alert-outbox", controllers.ListAlertOutbox) api.POST("/alert-outbox/:id/retry", controllers.RetryAlertOutbox) diff --git a/internal/systemsettings/sync.go b/internal/systemsettings/sync.go new file mode 100644 index 0000000..543c14f --- /dev/null +++ b/internal/systemsettings/sync.go @@ -0,0 +1,71 @@ +package systemsettings + +import ( + "context" + "log" + "sync/atomic" + "time" + + "git.apinb.com/ops/logs/internal/config" + "git.apinb.com/ops/logs/internal/impl" + "git.apinb.com/ops/logs/internal/models" + settingsv1 "git.apinb.com/ops/pkgs/systemsettings/v1" +) + +var retentionDays atomic.Int64 + +func LoadInitial(ctx context.Context) (*settingsv1.Client, error) { + client, err := settingsv1.NewClient( + config.Spec.SystemSettings.BaseURL, + time.Duration(config.Spec.SystemSettings.TimeoutSeconds)*time.Second, + ) + if err != nil { + return nil, err + } + values, err := client.Fetch(ctx) + if err != nil { + return nil, err + } + value, err := values.Require(settingsv1.OperationLogRetentionDays) + if err != nil { + return nil, err + } + retentionDays.Store(int64(value)) + return client, nil +} + +func Start(ctx context.Context, client *settingsv1.Client) { + refreshTicker := time.NewTicker(time.Duration(config.Spec.SystemSettings.RefreshSeconds) * time.Second) + cleanupTicker := time.NewTicker(time.Hour) + defer refreshTicker.Stop() + defer cleanupTicker.Stop() + cleanup(ctx) + for { + select { + case <-ctx.Done(): + return + case <-refreshTicker.C: + values, err := client.Fetch(ctx) + if err != nil { + log.Printf("logs: 刷新系统参数失败: %v", err) + continue + } + value, err := values.Require(settingsv1.OperationLogRetentionDays) + if err != nil { + log.Printf("logs: 应用系统参数失败: %v", err) + continue + } + retentionDays.Store(int64(value)) + case <-cleanupTicker.C: + cleanup(ctx) + } + } +} + +func cleanup(ctx context.Context) { + cutoff := time.Now().UTC().AddDate(0, 0, -int(retentionDays.Load())) + result := impl.DBService.WithContext(ctx).Where("created_at < ?", cutoff).Delete(&models.AuditLog{}) + if result.Error != nil { + log.Printf("logs: 清理过期操作日志失败: %v", result.Error) + } +}