Logan.Li
f3a4fb569f
- 0004_metric_aggregate_unique 改为幂等：新库基线已由模型 uniqueIndex 建出 uniq_metric_agg，再次 CREATE 会以 1061 重复键名中断 migrate up， 导致 0005_agent_settings 建不出来（全新部署实测复现） - 新增 database.IndexExists：跨 SQLite/MySQL/PostgreSQL 判断索引是否存在 - 新增 database.EnsureSchemaInitialized：服务启动遇到空库时明确失败并给出 执行 aiops migrate up 的指引，避免「服务起来了但页面全是空数据」 - run.sh docker up/upgrade/update 先跑一次 aiops migrate up 再启动服务 // Logan.Li
390 lines
15 KiB
Go
390 lines
15 KiB
Go
package database
|
||
|
||
import (
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"ezops/config"
|
||
"ezops/model"
|
||
"ezops/pkg/appdir"
|
||
|
||
"github.com/glebarez/sqlite"
|
||
"gorm.io/driver/mysql"
|
||
"gorm.io/driver/postgres"
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/logger"
|
||
"gorm.io/gorm/schema"
|
||
)
|
||
|
||
// New 创建数据库连接
|
||
func New(cfg config.DatabaseConfig) (*gorm.DB, error) {
|
||
var dialector gorm.Dialector
|
||
|
||
switch cfg.Driver {
|
||
case "mysql":
|
||
dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=%s&parseTime=True&loc=Local",
|
||
cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.DBName, cfg.Charset)
|
||
dialector = mysql.Open(dsn)
|
||
case "postgres":
|
||
dsn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
|
||
cfg.Host, cfg.Port, cfg.User, cfg.Password, cfg.DBName)
|
||
dialector = postgres.Open(dsn)
|
||
case "sqlite":
|
||
dsn := appdir.Resolve(cfg.DBName)
|
||
if err := appdir.EnsureParentDir(dsn); err != nil {
|
||
return nil, fmt.Errorf("创建数据库目录失败: %w", err)
|
||
}
|
||
if !strings.Contains(dsn, "?") {
|
||
// WAL 提升并发读，busy_timeout 降低锁冲突 // Logan.Li
|
||
dsn += "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)"
|
||
}
|
||
dialector = sqlite.Open(dsn)
|
||
if cfg.MaxConn > 4 {
|
||
cfg.MaxConn = 4
|
||
}
|
||
if cfg.MaxIdle > 2 {
|
||
cfg.MaxIdle = 2
|
||
}
|
||
default:
|
||
return nil, fmt.Errorf("unsupported database driver: %s", cfg.Driver)
|
||
}
|
||
|
||
db, err := gorm.Open(dialector, &gorm.Config{
|
||
Logger: logger.Default.LogMode(logger.Silent),
|
||
NamingStrategy: indexNamedNaming{},
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 设置连接池
|
||
sqlDB, err := db.DB()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
sqlDB.SetMaxOpenConns(cfg.MaxConn)
|
||
sqlDB.SetMaxIdleConns(cfg.MaxIdle)
|
||
sqlDB.SetConnMaxLifetime(time.Hour)
|
||
|
||
return db, nil
|
||
}
|
||
|
||
// indexNamedNaming 让唯一索引与普通索引使用同一套命名规则（idx_ 前缀）。
|
||
// GORM 1.30 的 AutoMigrate 在表已存在时会对唯一索引列执行 `DROP INDEX uni_<表>_<列>`
|
||
// （NamingStrategy.UniqueName 用 uni_ 前缀），而建表实际使用 `idx_<表>_<列>`，
|
||
// 索引名永远不匹配，空库执行基线必然以 1091 失败。 // Logan.Li
|
||
type indexNamedNaming struct {
|
||
schema.NamingStrategy
|
||
}
|
||
|
||
// UniqueName 保持与 IndexName 一致，避免 AutoMigrate 误删不存在的唯一索引。
|
||
func (n indexNamedNaming) UniqueName(table, column string) string {
|
||
return n.NamingStrategy.IndexName(table, column)
|
||
}
|
||
|
||
// EnsureSchemaInitialized 校验库结构是否已初始化（即已执行过 `aiops migrate up`）。
|
||
// 服务启动不做结构迁移，但遇到空库必须明确失败，避免「服务已起来、页面全是空数据」的静默故障。 // Logan.Li
|
||
func EnsureSchemaInitialized(db *gorm.DB) error {
|
||
if db == nil {
|
||
return fmt.Errorf("数据库连接为空")
|
||
}
|
||
if !db.Migrator().HasTable("iam_users") {
|
||
return fmt.Errorf("未检测到业务表，数据库尚未初始化")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// BootstrapSchema 为空数据库创建初始基线结构。
|
||
// 该函数只能由显式的 `aiops migrate up` 调用，服务启动路径禁止调用。 // Logan.Li
|
||
func BootstrapSchema(db *gorm.DB) error {
|
||
allModels := []interface{}{
|
||
&model.User{},
|
||
&model.Role{},
|
||
&model.Permission{},
|
||
&model.Server{},
|
||
&model.Cloud{},
|
||
&model.CloudAccount{},
|
||
&model.Region{},
|
||
&model.Domain{},
|
||
&model.KubernetesCluster{},
|
||
&model.KubernetesNode{},
|
||
&model.KubernetesPod{},
|
||
&model.KubernetesDeployment{},
|
||
&model.KubernetesService{},
|
||
&model.KubernetesIngress{},
|
||
&model.AlertRule{},
|
||
&model.AlertEvent{},
|
||
&model.AlertContact{}, // 报警联系人 // Logan.Li
|
||
&model.AlertContactGroup{}, // 报警联系组 // Logan.Li
|
||
&model.AlertNotifyPolicy{}, // 告警集合通知策略 // Logan.Li
|
||
&model.AlertNotifySettings{}, // 告警消息模版 // Logan.Li
|
||
&model.NotifyChannel{},
|
||
&model.Pipeline{},
|
||
&model.PipelineRun{},
|
||
&model.JenkinsConfig{},
|
||
&model.BuildNode{},
|
||
&model.AppRelease{},
|
||
&model.CronJob{},
|
||
&model.Script{},
|
||
&model.TaskExecution{},
|
||
&model.ApprovalTemplate{},
|
||
&model.ApprovalProcess{},
|
||
&model.ApprovalStep{},
|
||
&model.NavigationGroup{},
|
||
&model.NavigationLink{},
|
||
&model.AuditLog{},
|
||
&model.AuditSettings{}, // 审计保留策略 // Logan.Li
|
||
&model.LoginLog{}, // 登录日志 // Logan.Li
|
||
&model.SystemTaskStatus{}, // 内置系统任务状态 // Logan.Li
|
||
&model.LDAPConfig{},
|
||
&model.ResourceModel{},
|
||
&model.CloudBilling{},
|
||
&model.BillingSettings{}, // 云账单换算设置 // Logan.Li
|
||
&model.BillingAlertRule{},
|
||
&model.AgentReport{},
|
||
&model.AgentTask{},
|
||
&model.AgentRule{},
|
||
&model.AgentAlert{},
|
||
&model.AgentPortPolicy{}, // Agent 端口策略 // Logan.Li
|
||
&model.AnsiblePlaybook{},
|
||
&model.AnsibleJob{},
|
||
&model.AnsibleInventory{},
|
||
&model.TopologyLayout{}, // 拓扑布局保存 // Logan.Li
|
||
&model.TerminalSession{}, // 终端会话录制（终端审计） // Logan.Li
|
||
&model.RDPSession{}, // RDP 图形会话审计 // Logan.Li
|
||
&model.SessionCommand{}, // 终端会话命令审计 // Logan.Li
|
||
&model.LLMConfig{}, // 大语言模型配置（AI 智能助手） // Logan.Li
|
||
&model.ChatHistory{}, // 对话历史记录（按用户隔离） // Logan.Li
|
||
&model.AIAnalysisTrace{}, // AI 分析上下文与结果审计回放 // Logan.Li
|
||
&model.MetricsAggregate{}, // 指标时序聚合（Redis → MySQL）// Logan.Li
|
||
&model.MonitorTarget{}, // 监控目标（主机/中间件/设备/站点） // Logan.Li
|
||
&model.MonitorMetric{}, // 监控指标定义 // Logan.Li
|
||
&model.MetricHistory{}, // 指标历史时序 // Logan.Li
|
||
&model.HostGroup{}, // 主机组 // Logan.Li
|
||
&model.SNMPDevice{}, // SNMP设备 // Logan.Li
|
||
&model.SNMPTemplate{}, // SNMP模板 // Logan.Li
|
||
&model.SNMPTemplateOID{}, // SNMP模板OID // Logan.Li
|
||
&model.SNMPMetric{}, // SNMP采集指标 // Logan.Li
|
||
&model.SNMPInterface{}, // SNMP接口信息 // Logan.Li
|
||
&model.SiteTask{}, // 站点拨测任务 // Logan.Li
|
||
&model.SiteResult{}, // 拨测结果 // Logan.Li
|
||
&model.SiteAlert{}, // 拨测告警 // Logan.Li
|
||
&model.MonitorAlertRule{}, // 中间件/站点拨测告警规则 // Logan.Li
|
||
&model.PrometheusDataSource{}, // Prometheus 数据源配置 // Logan.Li
|
||
&model.ZabbixDataSource{}, // Zabbix 数据源配置 // Logan.Li
|
||
&model.LogDataSource{}, // Loki / ES 日志数据源 // Logan.Li
|
||
&model.Credential{}, // 凭据管理（SSH密钥/API Token） // Logan.Li
|
||
&model.DeployEnvironment{}, // 发布环境分组 // Logan.Li
|
||
&model.DeployPipelineMember{}, // 流水线成员 // Logan.Li
|
||
&model.DeployUserGroup{}, // 用户组 // Logan.Li
|
||
&model.DeployUserGroupMember{}, // 用户组成员 // Logan.Li
|
||
&model.ItemRole{}, // Jenkins-style Item Role (regex pipeline matching) // Logan.Li
|
||
&model.UserItemRole{}, // User ↔ ItemRole association // Logan.Li
|
||
&model.LogScopeRole{}, // 日志查询范围权限（正则匹配索引/Label）// Logan.Li
|
||
&model.UserLogScopeRole{}, // 用户 ↔ 日志范围权限 // Logan.Li
|
||
&model.BuildMasterConfig{}, // Master 构建节点并发配置 // Logan.Li
|
||
&model.DeployTool{}, // 工具配置（懒加载/预装/禁用） // Logan.Li
|
||
&model.AlertCondition{}, // 告警条件 // Logan.Li
|
||
&model.AlertNotifyTemplate{}, // 告警通知模板 // Logan.Li
|
||
&model.AlertUpgrade{}, // 告警升级策略 // Logan.Li
|
||
&model.AlertUpgradeRecord{}, // 告警升级记录 // Logan.Li
|
||
&model.AlertSilence{}, // 告警静默 // Logan.Li
|
||
&model.HealthProbe{}, // 健康探针配置 // Logan.Li
|
||
&model.ApprovalWorkflow{}, // 审批流程定义 // Logan.Li
|
||
&model.ApprovalWorkflowStep{}, // 审批流程步骤 // Logan.Li
|
||
&model.AIRun{}, // AI 任务统一运行记录 // Logan.Li
|
||
&model.AIRunMessage{}, // AI 任务回放消息 // Logan.Li
|
||
&model.AIRunSpan{}, // AI 工具/LLM/Guardrail Trace Span // Logan.Li
|
||
&model.AICapability{}, // AI 内置/MCP/技能能力注册 // Logan.Li
|
||
&model.AICapabilityPolicy{}, // AI 工具能力策略 // Logan.Li
|
||
&model.OpsSession{}, // 人机统一执行会话 // Logan.Li
|
||
&model.OpsCommand{}, // 人机统一命令审计 // Logan.Li
|
||
&model.TraceDataSource{}, // 链路追踪数据源 Jaeger/Tempo // Logan.Li
|
||
}
|
||
|
||
// 必须一次性传入全部模型：GORM 只在单次调用内做外键依赖排序
|
||
// （如 asset_cloud_regions 必须在 asset_servers 之前建表），逐个建表会踩外键顺序问题。 // Logan.Li
|
||
if err := db.AutoMigrate(allModels...); err != nil {
|
||
return fmt.Errorf("创建初始基线表失败: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ApplyLegacySchemaPatches 为历史数据库执行一次兼容性补丁。
|
||
// 调用方必须通过版本化迁移显式执行，任一失败都会中止，避免隐式部分成功。 // Logan.Li
|
||
func ApplyLegacySchemaPatches(db *gorm.DB) error {
|
||
m := db.Migrator()
|
||
if !m.HasTable(&model.User{}) {
|
||
return fmt.Errorf("缺少 iam_users，不能执行历史结构补丁")
|
||
}
|
||
patchUserColumn := func(field string, col string) error {
|
||
if !m.HasColumn(&model.User{}, col) {
|
||
if err := m.AddColumn(&model.User{}, field); err != nil {
|
||
return fmt.Errorf("补充 iam_users.%s 失败: %w", col, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
for _, column := range [][2]string{
|
||
{"ApiAccessKeyExpiresAt", "api_access_key_expires_at"},
|
||
{"ApiAccessKeyID", "api_access_key_id"},
|
||
{"ApiAccessKeyDescription", "api_access_key_description"},
|
||
{"ApiAccessKeyDurationMonths", "api_access_key_duration_months"},
|
||
{"ApiAccessKeyLastUsedAt", "api_access_key_last_used_at"},
|
||
} {
|
||
if err := patchUserColumn(column[0], column[1]); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
// 补全日志数据源表可能缺失的列（旧表仅有基础字段）// Logan.Li
|
||
patchLogDSColumn := func(field string, col string) error {
|
||
if !m.HasColumn(&model.LogDataSource{}, col) {
|
||
if err := m.AddColumn(&model.LogDataSource{}, field); err != nil {
|
||
return fmt.Errorf("补充 monitor_log_datasources.%s 失败: %w", col, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
for _, column := range [][2]string{{"AuthType", "auth_type"}, {"Token", "token"}, {"Headers", "headers"}} {
|
||
if err := patchLogDSColumn(column[0], column[1]); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
// 日志权限关联表：AutoMigrate 在 MySQL 索引冲突时可能只创建策略表未创建用户关联表 // Logan.Li
|
||
ensureTable := func(tbl interface{}, name string) error {
|
||
if !m.HasTable(tbl) {
|
||
if err := db.AutoMigrate(tbl); err != nil {
|
||
return fmt.Errorf("创建 %s 失败: %w", name, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
if err := ensureTable(&model.LogScopeRole{}, "iam_log_scope_roles"); err != nil {
|
||
return err
|
||
}
|
||
if !m.HasTable("iam_user_log_scope_roles") {
|
||
if err := createUserLogScopeRolesTable(db); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
// LDAP 目录类型列（openldap | ad） // Logan.Li
|
||
if m.HasTable(&model.LDAPConfig{}) && !m.HasColumn(&model.LDAPConfig{}, "server_type") {
|
||
if err := m.AddColumn(&model.LDAPConfig{}, "ServerType"); err != nil {
|
||
return fmt.Errorf("补充 iam_ldap_configs.server_type 失败: %w", err)
|
||
}
|
||
}
|
||
|
||
// 服务器 RDP/协议字段（旧库 AutoMigrate 可能未加列）// Logan.Li
|
||
patchServerColumn := func(field string, col string) error {
|
||
if !m.HasColumn(&model.Server{}, col) {
|
||
if err := m.AddColumn(&model.Server{}, field); err != nil {
|
||
return fmt.Errorf("补充 asset_servers.%s 失败: %w", col, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
for _, column := range [][2]string{
|
||
{"OSType", "os_type"}, {"Protocol", "protocol"}, {"RDPPort", "rdp_port"},
|
||
{"Alias", "alias"}, {"CloudInstanceName", "cloud_instance_name"}, {"CloudInstanceID", "cloud_instance_id"},
|
||
} {
|
||
if err := patchServerColumn(column[0], column[1]); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
// 将历史仅填 os=windows 的记录补齐协议字段 // Logan.Li
|
||
if m.HasColumn(&model.Server{}, "os_type") && m.HasColumn(&model.Server{}, "protocol") {
|
||
if err := db.Exec(`
|
||
UPDATE asset_servers
|
||
SET os_type = 'windows', protocol = 'rdp', rdp_port = CASE WHEN rdp_port IS NULL OR rdp_port = 0 THEN 3389 ELSE rdp_port END
|
||
WHERE (os_type IS NULL OR os_type = '' OR protocol IS NULL OR protocol = '')
|
||
AND LOWER(os) LIKE '%windows%'
|
||
`).Error; err != nil {
|
||
return fmt.Errorf("补全 Windows 服务器协议字段失败: %w", err)
|
||
}
|
||
}
|
||
|
||
// 兼容早期 GORM 将 CIProvider 误建为 c_iprovider 的流水线表。// Logan.Li
|
||
if m.HasTable(&model.Pipeline{}) {
|
||
if !m.HasColumn(&model.Pipeline{}, "ci_provider") {
|
||
if err := m.AddColumn(&model.Pipeline{}, "CIProvider"); err != nil {
|
||
return fmt.Errorf("补充 deploy_pipelines.ci_provider 失败: %w", err)
|
||
}
|
||
}
|
||
if m.HasColumn(&model.Pipeline{}, "ci_provider") && m.HasColumn(&model.Pipeline{}, "c_iprovider") {
|
||
if err := db.Exec(`
|
||
UPDATE deploy_pipelines
|
||
SET ci_provider = COALESCE(NULLIF(ci_provider, ''), NULLIF(c_iprovider, ''), 'aiops')
|
||
WHERE ci_provider IS NULL OR ci_provider = ''
|
||
`).Error; err != nil {
|
||
return fmt.Errorf("迁移 deploy_pipelines.ci_provider 失败: %w", err)
|
||
}
|
||
}
|
||
patchPipelineColumn := func(field string, col string) error {
|
||
if !m.HasColumn(&model.Pipeline{}, col) {
|
||
if err := m.AddColumn(&model.Pipeline{}, field); err != nil {
|
||
return fmt.Errorf("补充 deploy_pipelines.%s 失败: %w", col, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
for _, column := range [][2]string{{"JenkinsView", "jenkins_view"}, {"JenkinsViewURL", "jenkins_view_url"}, {"JenkinsViews", "jenkins_views"}, {"JenkinsParams", "jenkins_params"}, {"JenkinsSteps", "jenkins_steps"}} {
|
||
if err := patchPipelineColumn(column[0], column[1]); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
if m.HasTable(&model.JenkinsConfig{}) {
|
||
patchJenkinsColumn := func(field string, col string) error {
|
||
if !m.HasColumn(&model.JenkinsConfig{}, col) {
|
||
if err := m.AddColumn(&model.JenkinsConfig{}, field); err != nil {
|
||
return fmt.Errorf("补充 deploy_jenkins_configs.%s 失败: %w", col, err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
if err := patchJenkinsColumn("RefMode", "ref_mode"); err != nil {
|
||
return err
|
||
}
|
||
if err := patchJenkinsColumn("RefParam", "ref_param"); err != nil {
|
||
return err
|
||
}
|
||
if err := db.Exec("UPDATE deploy_jenkins_configs SET ref_mode = 'both' WHERE ref_mode IS NULL OR ref_mode = ''").Error; err != nil {
|
||
return fmt.Errorf("补全 deploy_jenkins_configs.ref_mode 失败: %w", err)
|
||
}
|
||
if err := db.Exec("UPDATE deploy_jenkins_configs SET ref_param = 'SELECT' WHERE ref_param IS NULL OR ref_param = ''").Error; err != nil {
|
||
return fmt.Errorf("补全 deploy_jenkins_configs.ref_param 失败: %w", err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func createUserLogScopeRolesTable(db *gorm.DB) error {
|
||
sql := `CREATE TABLE IF NOT EXISTS iam_user_log_scope_roles (
|
||
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
|
||
user_id BIGINT UNSIGNED NOT NULL,
|
||
log_scope_role_id BIGINT UNSIGNED NOT NULL,
|
||
created_at DATETIME(3) NULL,
|
||
UNIQUE KEY idx_iam_user_log_scope_role (user_id, log_scope_role_id),
|
||
KEY idx_iam_user_log_scope_role_role (log_scope_role_id)
|
||
)`
|
||
if db.Dialector.Name() == "sqlite" {
|
||
sql = `CREATE TABLE IF NOT EXISTS iam_user_log_scope_roles (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id INTEGER NOT NULL,
|
||
log_scope_role_id INTEGER NOT NULL,
|
||
created_at DATETIME,
|
||
UNIQUE(user_id, log_scope_role_id)
|
||
)`
|
||
}
|
||
if err := db.Exec(sql).Error; err != nil {
|
||
return fmt.Errorf("创建 iam_user_log_scope_roles 失败: %w", err)
|
||
}
|
||
return nil
|
||
}
|