Files
aiops/backend/main.go
T

544 lines
16 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package main
import (
"context"
"flag"
"fmt"
"net/http"
"os"
"os/signal"
"path/filepath"
"strings"
"syscall"
"time"
"ezops/config"
"ezops/controller"
"ezops/model"
"ezops/pkg/appdir"
"ezops/pkg/auditcleaner"
"ezops/pkg/billingsync"
"ezops/pkg/cronscheduler"
"ezops/pkg/database"
"ezops/pkg/logger"
ezredis "ezops/pkg/redis"
snmppkg "ezops/pkg/snmp"
"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, "跳过启动时的数据库自动迁移(仅用于已完成迁移的故障恢复)")
// 数据库参数
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
// @host localhost:8080
// @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)
}
}
// 解析 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))
}
// 自动迁移;大 SQLite 库异常迁移时可显式跳过,先恢复已存在的服务。 // Logan.Li
if *flagSkipAutoMigrate {
log.Warn("已跳过数据库自动迁移,仅限已完成迁移的故障恢复场景")
} else if err := database.AutoMigrate(db); err != nil {
log.Fatal("数据库迁移失败", zap.Error(err))
}
// 初始化种子数据
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 上报数据(保留 7 天,防止表膨胀影响查询性能)// Logan.Li
go func() {
ticker := time.NewTicker(24 * time.Hour)
defer ticker.Stop()
runAgentCleanup := func() {
start := time.Now()
cleanupSQL := database.CleanupAgentReportsSQL(cfg.Database.Driver)
r := db.Exec(cleanupSQL)
msg := fmt.Sprintf("清理 %d 条记录", r.RowsAffected)
status := "success"
if r.Error != nil {
status = "failed"
msg = r.Error.Error()
log.Warn("Agent 上报清理失败", zap.Error(r.Error))
} else if r.RowsAffected > 0 {
log.Info("清理过期的 agent 上报数据", zap.Int64("rows", r.RowsAffected))
}
systask.Record("agent_report_cleanup", status, msg, time.Since(start))
}
time.Sleep(2 * time.Minute)
runAgentCleanup()
for range ticker.C {
runAgentCleanup()
}
}()
// 定期清理超期 SNMP 采集指标(保留 7 天) // Logan.Li
go func() {
ticker := time.NewTicker(24 * time.Hour)
defer ticker.Stop()
runSNMPCleanup := func() {
start := time.Now()
cleanupSQL := database.CleanupSNMPMetricsSQL(cfg.Database.Driver, snmppkg.MetricsRetentionDays)
r := db.Exec(cleanupSQL)
msg := fmt.Sprintf("清理 %d 条 SNMP 指标(保留 %d 天)", r.RowsAffected, snmppkg.MetricsRetentionDays)
status := "success"
if r.Error != nil {
status = "failed"
msg = r.Error.Error()
log.Warn("SNMP 指标清理失败", zap.Error(r.Error))
} else if r.RowsAffected > 0 {
log.Info("清理过期的 SNMP 指标", zap.Int64("rows", r.RowsAffected))
}
systask.Record("snmp_metrics_cleanup", status, msg, time.Since(start))
}
time.Sleep(3 * time.Minute)
runSNMPCleanup()
for range ticker.C {
runSNMPCleanup()
}
}()
// 每日按审计保留策略清理操作/登录/终端审计日志 // 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("服务已退出")
}
// 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("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()
}
}