package main import ( "context" "encoding/json" "flag" "fmt" "net/http" "os" "os/signal" "path/filepath" "strings" "syscall" "time" "ezops/config" "ezops/controller" "ezops/migrations" "ezops/model" "ezops/pkg/appdir" "ezops/pkg/auditcleaner" "ezops/pkg/billingsync" "ezops/pkg/cronscheduler" "ezops/pkg/database" "ezops/pkg/logger" ezredis "ezops/pkg/redis" "ezops/pkg/retention" "ezops/pkg/systask" "ezops/router" "ezops/seed" "github.com/gin-gonic/gin" "github.com/spf13/viper" "go.uber.org/zap" ) // 版本信息(通过编译时 ldflags 注入)// Logan.Li var ( Version = "dev" BuildTime = "unknown" GitCommit = "unknown" ) // CLI 参数定义 // Logan.Li var ( flagPort = flag.String("port", "", "服务端口 (默认: 8080)") flagConfig = flag.String("config", "", "配置文件路径 (默认: config.yaml)") flagMode = flag.String("mode", "", "运行模式: local (SQLite+redka), development, production, release") flagLogLevel = flag.String("log-level", "", "日志级别: debug, info, warn, error") flagSkipAutoMigrate = flag.Bool("skip-auto-migrate", false, "已废弃:服务启动不再执行自动迁移") // Logan.Li // 数据库参数 flagDBDriver = flag.String("db-driver", "", "数据库驱动: sqlite, mysql, postgres (默认: sqlite)") flagDBHost = flag.String("db-host", "", "数据库主机") flagDBPort = flag.Int("db-port", 0, "数据库端口") flagDBUser = flag.String("db-user", "", "数据库用户名") flagDBPassword = flag.String("db-password", "", "数据库密码") flagDBName = flag.String("db-name", "", "数据库名 / SQLite 文件路径") flagDBCharset = flag.String("db-charset", "", "数据库字符集") // Redis 参数 flagRedisHost = flag.String("redis-host", "", "Redis 主机") flagRedisPort = flag.Int("redis-port", 0, "Redis 端口") flagRedisPassword = flag.String("redis-password", "", "Redis 密码") flagRedisDB = flag.Int("redis-db", 0, "Redis 数据库编号") flagRedisDriver = flag.String("redis-driver", "", "Redis 驱动: redis (外部Redis), redka (内嵌,需CGO)") // JWT 参数 flagJWTSecret = flag.String("jwt-secret", "", "JWT 密钥") flagJWTExpire = flag.Int("jwt-expire", 0, "JWT 过期时间 (小时)") // 显示版本 flagVersion = flag.Bool("version", false, "显示版本信息") ) // @title AIOps运维平台 API // @version 1.0 // @description AIOps智能运维平台后端API文档 // @termsOfService http://swagger.io/terms/ // @contact.name API Support // @contact.url http://www.swagger.io/support // @contact.email support@swagger.io // @license.name Apache 2.0 // @license.url http://www.apache.org/licenses/LICENSE-2.0.html // @BasePath /api/v1 // @securityDefinitions.apikey Bearer // @in header // @name Authorization // @description Type "Bearer" followed by a space and JWT token. // printVersion 输出编译时注入的版本信息 // Logan.Li func printVersion() { fmt.Printf("AIOps 智能运维平台\n") fmt.Printf(" 版本: %s\n", Version) fmt.Printf(" 构建: %s\n", BuildTime) fmt.Printf(" 提交: %s\n", GitCommit) } func main() { // 支持 ./aiops version 子命令(无需启动服务)// Logan.Li if len(os.Args) > 1 { switch os.Args[1] { case "version", "-v", "--version": printVersion() os.Exit(0) case "migrate": runMigrationCLI(os.Args[2:]) return } } // 解析 CLI 参数 // Logan.Li flag.Parse() // 显示版本信息 // Logan.Li if *flagVersion { printVersion() os.Exit(0) } // 加载配置 // Logan.Li cfg := loadConfig() // 初始化日志 log := logger.New(cfg.LogLevel) defer log.Sync() log.Info("AIOps 启动中", zap.String("version", Version), zap.String("mode", cfg.Mode), zap.String("port", cfg.Port), ) // 设置 Gin 模式(local 与 release 一样走 ReleaseMode)// Logan.Li switch cfg.Mode { case "production", "release", "local": gin.SetMode(gin.ReleaseMode) default: gin.SetMode(gin.DebugMode) } // 初始化数据库 db, err := database.New(cfg.Database) if err != nil { log.Fatal("数据库连接失败", zap.Error(err)) } // 数据库结构只能通过 `aiops migrate up` 显式变更,避免启动期重建表和 WAL 写放大。 // Logan.Li if *flagSkipAutoMigrate { log.Warn("--skip-auto-migrate 已废弃;服务启动始终不会执行数据库迁移") } // 初始化种子数据 seed.SeedData(db) // 中间件监控默认目标(aiops-redis / aiops-mysql)// Logan.Li controller.SeedDefaultMiddlewareTargets(db) // 初始化缓存(Redka 内嵌 / Redis 外部)// Logan.Li cache, err := ezredis.New(ezredis.Config{ Driver: cfg.Redis.Driver, // 从配置读取 driver Host: cfg.Redis.Host, Port: cfg.Redis.Port, Password: cfg.Redis.Password, DB: cfg.Redis.DB, DataDir: appdir.Root(), }) if err != nil { if cfg.Mode == "local" { log.Fatal("local 模式缓存(redka)初始化失败", zap.Error(err)) } log.Warn("缓存初始化失败,继续启动", zap.Error(err)) cache = nil } else { caps := cache.Capabilities() log.Info("缓存已就绪", zap.String("driver", cfg.Redis.Driver), zap.Bool("sorted_set", caps.SortedSet), zap.Bool("pubsub", caps.PubSub), ) if cfg.Mode == "local" && (!caps.SortedSet || !caps.PubSub) { log.Info("local 模式:Agent 热指标走 DB 直读,流水线日志走轮询") } defer cache.Close() } // 创建Gin实例 r := gin.New() // 中间件 r.Use(gin.Recovery()) r.Use(logger.GinLogger(log)) r.Use(CORS()) // 版本信息 API r.GET("/version", func(c *gin.Context) { resp := gin.H{ "version": Version, "build_time": BuildTime, "git_commit": GitCommit, "mode": cfg.Mode, "db_driver": cfg.Database.Driver, "redis_driver": cfg.Redis.Driver, } if cache != nil { resp["features"] = cache.Capabilities() } c.JSON(http.StatusOK, resp) }) systask.Init(db) // 注册路由 router.Register(r, db, log, cfg, cache) // 用户定时任务调度(Linux 5 段 crontab)// Logan.Li cronSched := cronscheduler.Start(db, func(jobID uint) { if _, err := controller.RunCronJobByID(db, jobID); err != nil { log.Warn("定时任务自动执行失败", zap.Uint("job_id", jobID), zap.Error(err)) } }) defer cronSched.Stop() // 每日分批清理 Agent、指标、SNMP、拨测与旧日志,执行结果写入系统任务状态。 // Logan.Li go func() { ticker := time.NewTicker(24 * time.Hour) defer ticker.Stop() runRetentionCleanup := func() { start := time.Now() result, err := retention.Run(db, cfg.Retention) if err != nil { systask.Record("retention_cleanup", "failed", err.Error(), time.Since(start)) log.Warn("高频数据保留清理失败", zap.Error(err)) return } message, marshalErr := json.Marshal(result) if marshalErr != nil { systask.Record("retention_cleanup", "failed", marshalErr.Error(), time.Since(start)) log.Warn("高频数据保留结果序列化失败", zap.Error(marshalErr)) return } systask.Record("retention_cleanup", "success", string(message), time.Since(start)) log.Info("高频数据保留清理完成", zap.Int64("rows", result.Total)) } time.Sleep(2 * time.Minute) runRetentionCleanup() for range ticker.C { runRetentionCleanup() } }() // 每日按审计保留策略清理操作/登录/终端审计日志 // Logan.Li go func() { ticker := time.NewTicker(24 * time.Hour) defer ticker.Stop() runAuditRetentionCleanup := func() { start := time.Now() result, err := auditcleaner.RunFromSettings(db) if err != nil { systask.Record("audit_log_cleanup", "failed", err.Error(), time.Since(start)) log.Warn("审计日志定时清理失败", zap.Error(err)) return } msg := fmt.Sprintf("清理 %d 条(保留 %d 天)", result.Total, result.RetentionDays) systask.Record("audit_log_cleanup", "success", msg, time.Since(start)) if result.Total > 0 { log.Info("审计日志定时清理完成", zap.Int("retention_days", result.RetentionDays), zap.Int64("total", result.Total), ) } now := time.Now() summary := result.SummaryJSON() db.Model(&model.AuditSettings{}).Where("id > 0").Updates(map[string]interface{}{ "last_cleanup_at": now, "last_cleanup_summary": summary, }) } time.Sleep(time.Minute) runAuditRetentionCleanup() for range ticker.C { runAuditRetentionCleanup() } }() // 每日 10:30 同步云账号昨日账单(阿里云 T+1 出账) // Logan.Li go func() { runBillingSync := func() { start := time.Now() result := billingsync.SyncDate(db, billingsync.Yesterday(), nil) msg := billingsync.SummaryMessage(result) status := "success" if result.Accounts == 0 { status = "failed" msg = "没有可同步的活跃云账号" } else if result.Created == 0 && result.Updated == 0 && len(result.Errors) > 0 { status = "failed" msg = result.Errors[0] } else if len(result.Errors) > 0 { msg += ";" + result.Errors[0] } systask.Record("cloud_billing_sync", status, msg, time.Since(start)) if status == "success" && (result.Created > 0 || result.Updated > 0) { log.Info("云账单定时同步完成", zap.String("date", result.BillingDate), zap.Int("created", result.Created), zap.Int("updated", result.Updated), ) } } for { now := time.Now() next := time.Date(now.Year(), now.Month(), now.Day(), 10, 30, 0, 0, now.Location()) if !now.Before(next) { next = next.Add(24 * time.Hour) } time.Sleep(time.Until(next)) runBillingSync() } }() // 前端静态文件服务(go:embed 模式,二进制部署用) // Logan.Li setupFrontend(r) // 创建HTTP服务器 srv := &http.Server{ Addr: ":" + cfg.Port, Handler: r, } // 优雅关闭 go func() { log.Info("服务已启动", zap.String("addr", srv.Addr)) if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatal("服务启动失败", zap.Error(err)) } }() // 等待中断信号 quit := make(chan os.Signal, 1) signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) <-quit log.Info("正在关闭服务...") ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := srv.Shutdown(ctx); err != nil { log.Fatal("强制关闭服务", zap.Error(err)) } log.Info("服务已退出") } // runMigrationCLI 执行独立的数据库迁移命令,不启动 HTTP 服务或定时任务。 // Logan.Li func runMigrationCLI(args []string) { action, flagArgs, err := parseMigrationArgs(args) if err != nil { fmt.Fprintf(os.Stderr, "迁移命令错误: %v\n用法: aiops migrate [全局参数]\n", err) os.Exit(2) } // 复用既有全局配置参数解析,确保 migrate 与服务使用同一数据库连接。 // Logan.Li os.Args = append([]string{os.Args[0]}, flagArgs...) if err := flag.CommandLine.Parse(flagArgs); err != nil { fmt.Fprintf(os.Stderr, "迁移参数错误: %v\n", err) os.Exit(2) } cfg := loadConfig() db, err := database.New(cfg.Database) if err != nil { fmt.Fprintf(os.Stderr, "数据库连接失败: %v\n", err) os.Exit(1) } switch action { case "up": applied, err := migrations.Up(db) if err != nil { fmt.Fprintf(os.Stderr, "迁移失败: %v\n", err) os.Exit(1) } if len(applied) == 0 { fmt.Println("迁移已是最新状态") return } fmt.Printf("已应用迁移: %s\n", strings.Join(applied, ", ")) case "baseline": if err := migrations.Baseline(db); err != nil { fmt.Fprintf(os.Stderr, "基线登记失败: %v\n", err) os.Exit(1) } fmt.Printf("已登记基线: %s(未修改业务表)\n", migrations.BaselineVersion) case "status": rows, err := migrations.Status(db) if err != nil { fmt.Fprintf(os.Stderr, "读取迁移状态失败: %v\n", err) os.Exit(1) } for _, row := range rows { status := "pending" appliedAt := "-" if row.Applied { status = "applied" appliedAt = row.AppliedAt.Format(time.RFC3339) } fmt.Printf("%-30s %-8s %-25s %s\n", row.Version, status, appliedAt, row.Description) } } } // parseMigrationArgs 提取动作并保留其他全局参数,允许 `migrate up --config ...`。 // Logan.Li func parseMigrationArgs(args []string) (string, []string, error) { if len(args) == 0 { return "", nil, fmt.Errorf("缺少动作") } var action string flagArgs := make([]string, 0, len(args)-1) for _, arg := range args { if action == "" && (arg == "up" || arg == "status" || arg == "baseline") { action = arg continue } flagArgs = append(flagArgs, arg) } if action == "" { return "", nil, fmt.Errorf("动作必须是 up、status 或 baseline") } return action, flagArgs, nil } // loadConfig 加载配置(CLI flags 优先于 config.yaml)// Logan.Li func loadConfig() *config.Config { viper.SetConfigName("config") viper.SetConfigType("yaml") viper.AddConfigPath("./configs") viper.AddConfigPath(".") viper.AddConfigPath(appdir.Root()) // 如果指定了配置文件路径,使用它 // Logan.Li if *flagConfig != "" { viper.SetConfigFile(*flagConfig) } // 默认值(独立二进制模式:SQLite + 无 Redis)// Logan.Li viper.SetDefault("mode", "release") viper.SetDefault("port", "8080") viper.SetDefault("log_level", "info") viper.SetDefault("database.driver", "sqlite") viper.SetDefault("database.host", "localhost") viper.SetDefault("database.port", 3306) viper.SetDefault("database.user", "") viper.SetDefault("database.password", "") viper.SetDefault("database.dbname", "sqlite3.db") viper.SetDefault("database.charset", "utf8mb4") viper.SetDefault("database.max_conn", 100) viper.SetDefault("database.max_idle", 10) viper.SetDefault("redis.host", "localhost") viper.SetDefault("redis.port", 6379) viper.SetDefault("redis.driver", "redka") viper.SetDefault("redis.password", "") viper.SetDefault("redis.db", 0) viper.SetDefault("jwt.secret", "") // 生产环境必须通过配置或 JWT_SECRET 环境变量设置 viper.SetDefault("jwt.expire_hour", 24) viper.SetDefault("agent.enforce_auth", false) viper.SetDefault("retention.agent_reports_days", 7) viper.SetDefault("retention.metric_history_days", 30) viper.SetDefault("retention.metric_aggregate_days", 365) viper.SetDefault("retention.snmp_samples_days", 30) viper.SetDefault("retention.site_results_days", 30) viper.SetDefault("retention.log_entries_days", 30) viper.SetDefault("retention.cleanup_batch_size", 10000) viper.SetDefault("websocket.read_buffer_size", 1024) viper.SetDefault("websocket.write_buffer_size", 1024) // 读取配置文件(不存在也不报错)// Logan.Li if err := viper.ReadInConfig(); err != nil { // 配置文件不存在是正常的(独立二进制模式) if _, ok := err.(viper.ConfigFileNotFoundError); !ok { fmt.Fprintf(os.Stderr, "警告: 读取配置文件失败: %v,使用默认配置\n", err) } } // 环境变量覆盖 // Logan.Li viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) viper.AutomaticEnv() // CLI flags 覆盖(最高优先级)// Logan.Li applyFlags() // mode=local:强制 SQLite + redka(CLI 显式传 --db-driver/--redis-driver 时可覆盖)// Logan.Li applyLocalMode() // 解析配置 // Logan.Li var cfg config.Config if err := viper.Unmarshal(&cfg); err != nil { fmt.Fprintf(os.Stderr, "配置解析失败: %v\n", err) os.Exit(1) } // SQLite / redka 相对路径统一解析到可执行文件目录 // Logan.Li if cfg.Database.Driver == "sqlite" && cfg.Database.DBName != "" && !filepath.IsAbs(cfg.Database.DBName) { cfg.Database.DBName = appdir.Resolve(cfg.Database.DBName) } return &cfg } // isFlagSet 判断 CLI 参数是否被显式传入 // Logan.Li func isFlagSet(name string) bool { found := false flag.Visit(func(f *flag.Flag) { if f.Name == name { found = true } }) return found } // applyLocalMode local 模式:不依赖外部 MySQL/Redis,使用 SQLite + redka // Logan.Li func applyLocalMode() { if viper.GetString("mode") != "local" { return } if !isFlagSet("db-driver") { viper.Set("database.driver", "sqlite") } if !isFlagSet("db-name") { viper.Set("database.dbname", "sqlite3.db") } if !isFlagSet("redis-driver") { viper.Set("redis.driver", "redka") } } // applyFlags 将 CLI 参数映射到 viper(覆盖 config.yaml 和默认值)// Logan.Li func applyFlags() { // 检查是否有任何 flag 被显式设置 // Logan.Li flag.Visit(func(f *flag.Flag) { switch f.Name { case "port": viper.Set("port", *flagPort) case "mode": viper.Set("mode", *flagMode) case "log-level": viper.Set("log_level", *flagLogLevel) case "db-driver": viper.Set("database.driver", *flagDBDriver) case "db-host": viper.Set("database.host", *flagDBHost) case "db-port": viper.Set("database.port", *flagDBPort) case "db-user": viper.Set("database.user", *flagDBUser) case "db-password": viper.Set("database.password", *flagDBPassword) case "db-name": viper.Set("database.dbname", *flagDBName) case "db-charset": viper.Set("database.charset", *flagDBCharset) case "redis-host": viper.Set("redis.host", *flagRedisHost) case "redis-port": viper.Set("redis.port", *flagRedisPort) case "redis-password": viper.Set("redis.password", *flagRedisPassword) case "redis-db": viper.Set("redis.db", *flagRedisDB) case "redis-driver": viper.Set("redis.driver", *flagRedisDriver) case "jwt-secret": viper.Set("jwt.secret", *flagJWTSecret) case "jwt-expire": viper.Set("jwt.expire_hour", *flagJWTExpire) } }) } // CORS 跨域中间件 func CORS() gin.HandlerFunc { return func(c *gin.Context) { c.Writer.Header().Set("Access-Control-Allow-Origin", "*") c.Writer.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, PATCH, OPTIONS") c.Writer.Header().Set("Access-Control-Allow-Headers", "Origin, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization") c.Writer.Header().Set("Access-Control-Expose-Headers", "Content-Length") c.Writer.Header().Set("Access-Control-Allow-Credentials", "true") if c.Request.Method == "OPTIONS" { c.AbortWithStatus(204) return } c.Next() } }