diff --git a/backend/controller/agent.go b/backend/controller/agent.go index ffd86f0..d23e405 100644 --- a/backend/controller/agent.go +++ b/backend/controller/agent.go @@ -1,6 +1,7 @@ package controller import ( + "context" "encoding/json" "fmt" "math" @@ -122,7 +123,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { return `` }() + `, hostname = CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END, - updated_at = NOW()` + updated_at = ?` args := []interface{}{ req.CPU.UsagePercent, req.CPU.IOWait, req.Memory.UsagePercent, diskAvg, req.CPU.CoreCount, memMB, diskGB, @@ -149,6 +150,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { args = append(args, req.System.Uptime) } } + args = append(args, time.Now()) updateSQL += " WHERE hostname = ? OR ip = ?" args = append(args, req.Hostname, matchIP) @@ -219,12 +221,10 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { }) } - // 写入 Redis 有序集合（热数据，保留7天） - if h.cache != nil { + // 写入 Redis 有序集合热数据（redka/local 无 ZSet，数据已在 monitor_agent_reports）// Logan.Li + if h.cache != nil && h.cache.Capabilities().SortedSet { timestamp := float64(reportTime.Unix()) - ttl := 7 * 24 * time.Hour // 7天 - - // 使用 reportTime.UnixNano() 确保 member 唯一（避免相同值碰撞） + ttl := 7 * 24 * time.Hour h.writeHotMetric(c, "metrics:cpu:"+req.Hostname, req.CPU.UsagePercent, timestamp, ttl, reportTime) h.writeHotMetric(c, "metrics:mem:"+req.Hostname, req.Memory.UsagePercent, timestamp, ttl, reportTime) h.writeHotMetric(c, "metrics:load1:"+req.Hostname, req.CPU.Load1, timestamp, ttl, reportTime) @@ -343,7 +343,7 @@ func (h *AgentHandler) GetHostHistory(c *gin.Context) { }) } -// AggregateMetrics 聚合 Redis 热数据到 MySQL klines +// AggregateMetrics 聚合指标到 monitor_metrics_aggregate（Redis ZSet 或 DB 直读）// Logan.Li // POST /api/v1/agent/aggregate func (h *AgentHandler) AggregateMetrics(c *gin.Context) { var req model.AggregateReq @@ -352,13 +352,101 @@ func (h *AgentHandler) AggregateMetrics(c *gin.Context) { return } - if h.cache == nil { - c.JSON(http.StatusServiceUnavailable, model.Response{Code: 503, Message: "Redis 缓存不可用"}) + var totalUpserted int64 + var err error + if h.cache != nil && h.cache.Capabilities().SortedSet { + totalUpserted, err = h.aggregateFromRedis(c.Request.Context(), req) + } else { + totalUpserted, err = h.aggregateFromReports(req) + } + if err != nil { + c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: err.Error()}) return } - // 支持的聚合间隔及对应的回溯时长 - intervals := []struct { + c.JSON(http.StatusOK, model.Response{ + Code: 200, + Message: fmt.Sprintf("聚合完成，写入/更新 %d 条记录", totalUpserted), + Data: gin.H{"upserted": totalUpserted, "source": aggregateSource(h.cache)}, + }) +} + +func aggregateSource(cache ezredis.Cache) string { + if cache != nil && cache.Capabilities().SortedSet { + return "redis" + } + return "database" +} + +func (h *AgentHandler) aggregateFromRedis(ctx context.Context, req model.AggregateReq) (int64, error) { + if h.cache == nil { + return 0, fmt.Errorf("Redis 缓存不可用") + } + intervals := aggregateIntervals() + redisKey := fmt.Sprintf("metrics:%s:%s", req.Metric, req.Hostname) + var total int64 + + for _, iv := range intervals { + lookback := iv.duration * 200 + now := time.Now() + since := now.Add(-lookback) + members, err := h.cache.ZRangeByScore(ctx, redisKey, + fmt.Sprintf("%d", since.Unix()), fmt.Sprintf("%d", now.Unix()+1)) + if err != nil || len(members) == 0 { + continue + } + buckets := make(map[int64][]float64) + for _, m := range members { + ts := int64(m.Score) + bucketStart := (ts / int64(iv.duration.Seconds())) * int64(iv.duration.Seconds()) + buckets[bucketStart] = append(buckets[bucketStart], parseMemberValue(m.Member)) + } + n, err := h.upsertMetricBuckets(req, iv.name, buckets) + if err != nil { + return total, err + } + total += n + } + return total, nil +} + +// aggregateFromReports local 模式：从 monitor_agent_reports 聚合（无需 Redis ZSet）// Logan.Li +func (h *AgentHandler) aggregateFromReports(req model.AggregateReq) (int64, error) { + intervals := aggregateIntervals() + var total int64 + + for _, iv := range intervals { + lookback := iv.duration * 200 + since := time.Now().Add(-lookback) + var reports []model.AgentReport + if err := h.db.Where("hostname = ? AND report_time >= ?", req.Hostname, since). + Order("report_time ASC").Find(&reports).Error; err != nil { + return total, err + } + if len(reports) == 0 { + continue + } + buckets := make(map[int64][]float64) + sec := int64(iv.duration.Seconds()) + for _, r := range reports { + ts := r.ReportTime.Unix() + bucketStart := (ts / sec) * sec + buckets[bucketStart] = append(buckets[bucketStart], metricValueFromReport(r, req.Metric)) + } + n, err := h.upsertMetricBuckets(req, iv.name, buckets) + if err != nil { + return total, err + } + total += n + } + return total, nil +} + +func aggregateIntervals() []struct { + name string + duration time.Duration +} { + return []struct { name string duration time.Duration }{ @@ -369,102 +457,84 @@ func (h *AgentHandler) AggregateMetrics(c *gin.Context) { {"6h", 6 * time.Hour}, {"1d", 24 * time.Hour}, } +} - redisKey := fmt.Sprintf("metrics:%s:%s", req.Metric, req.Hostname) - ctx := c.Request.Context() - var totalUpserted int64 +func metricValueFromReport(r model.AgentReport, metric string) float64 { + switch metric { + case "cpu": + return r.CPUUsage + case "mem": + return r.MemUsage + case "load1": + return r.Load1 + default: + return 0 + } +} - for _, iv := range intervals { - // 回溯窗口：取 interval × 200 倍，确保覆盖足够多的数据点 - lookback := iv.duration * 200 - now := time.Now() - since := now.Add(-lookback) - minScore := fmt.Sprintf("%d", since.Unix()) - maxScore := fmt.Sprintf("%d", now.Unix()+1) - - members, err := h.cache.ZRangeByScore(ctx, redisKey, minScore, maxScore) - if err != nil || len(members) == 0 { +func (h *AgentHandler) upsertMetricBuckets(req model.AggregateReq, interval string, buckets map[int64][]float64) (int64, error) { + var total int64 + for bucketTs, values := range buckets { + if len(values) == 0 { continue } - - // 按间隔分桶 - buckets := make(map[int64][]float64) // bucketTime (unix) → values - for _, m := range members { - ts := int64(m.Score) - bucketStart := (ts / int64(iv.duration.Seconds())) * int64(iv.duration.Seconds()) - val := parseMemberValue(m.Member) - buckets[bucketStart] = append(buckets[bucketStart], val) + maxV, minV, avg, p95 := calcAggregateStats(values) + agg := model.MetricsAggregate{ + Hostname: req.Hostname, + Metric: req.Metric, + Interval: interval, + Avg: math.Round(avg*100) / 100, + MaxV: math.Round(maxV*100) / 100, + MinV: math.Round(minV*100) / 100, + P95: math.Round(p95*100) / 100, + Count: len(values), + BucketTime: time.Unix(bucketTs, 0), } + result := h.db.Clauses(clause.OnConflict{ + Columns: []clause.Column{ + {Name: "hostname"}, + {Name: "metric"}, + {Name: "interval"}, + {Name: "bucket_time"}, + }, + DoUpdates: clause.AssignmentColumns([]string{ + "avg", "max_v", "min_v", "p95", "count", + }), + }).Create(&agg) + if result.Error != nil { + return total, result.Error + } + total++ + } + return total, nil +} - // 计算 max/min/avg/p95 - for bucketTs, values := range buckets { - if len(values) == 0 { - continue - } - maxV := values[0] - minV := values[0] - sum := 0.0 - - for _, v := range values { - sum += v - if v > maxV { - maxV = v - } - if v < minV { - minV = v - } - } - avg := sum / float64(len(values)) - - // 计算 P95（排序后取 95% 位置的值） - p95 := maxV - if len(values) >= 20 { - sorted := make([]float64, len(values)) - copy(sorted, values) - sort.Float64s(sorted) - idx := int(float64(len(sorted)) * 0.95) - if idx >= len(sorted) { - idx = len(sorted) - 1 - } - p95 = sorted[idx] - } - - agg := model.MetricsAggregate{ - Hostname: req.Hostname, - Metric: req.Metric, - Interval: iv.name, - Avg: math.Round(avg*100) / 100, - MaxV: math.Round(maxV*100) / 100, - MinV: math.Round(minV*100) / 100, - P95: math.Round(p95*100) / 100, - Count: len(values), - BucketTime: time.Unix(bucketTs, 0), - } - - // Upsert: hostname + metric + interval + bucket_time 唯一 - result := h.db.Clauses(clause.OnConflict{ - Columns: []clause.Column{ - {Name: "hostname"}, - {Name: "metric"}, - {Name: "interval"}, - {Name: "bucket_time"}, - }, - DoUpdates: clause.AssignmentColumns([]string{ - "avg", "max_v", "min_v", "p95", "count", - }), - }).Create(&agg) - - if result.Error == nil { - totalUpserted++ - } +func calcAggregateStats(values []float64) (maxV, minV, avg, p95 float64) { + maxV = values[0] + minV = values[0] + sum := 0.0 + for _, v := range values { + sum += v + if v > maxV { + maxV = v + } + if v < minV { + minV = v } } - - c.JSON(http.StatusOK, model.Response{ - Code: 200, - Message: fmt.Sprintf("聚合完成，写入/更新 %d 条记录", totalUpserted), - Data: gin.H{"upserted": totalUpserted}, - }) + avg = sum / float64(len(values)) + p95 = maxV + if len(values) >= 20 { + sorted := make([]float64, len(values)) + copy(sorted, values) + sort.Float64s(sorted) + idx := int(float64(len(sorted)) * 0.95) + if idx >= len(sorted) { + idx = len(sorted) - 1 + } + p95 = sorted[idx] + } + return } // GetAggregate 查询聚合数据（原 GetKline） diff --git a/backend/controller/dashboard.go b/backend/controller/dashboard.go index 2c1bd93..808b94f 100644 --- a/backend/controller/dashboard.go +++ b/backend/controller/dashboard.go @@ -10,6 +10,7 @@ import ( "time" "ezops/model" + "ezops/pkg/database" ezredis "ezops/pkg/redis" "github.com/gin-gonic/gin" @@ -39,16 +40,18 @@ func (h *DashboardHandler) GetStats(c *gin.Context) { const cacheKey = "dashboard:stats" const cacheTTL = 30 * time.Second - // 尝试从 Redis 缓存读取 - if cached, err := h.cache.Get(c.Request.Context(), cacheKey); err == nil && cached != "" { - var stats model.DashboardStats - if json.Unmarshal([]byte(cached), &stats) == nil { - c.JSON(http.StatusOK, model.Response{ - Code: 200, - Message: "success", - Data: stats, - }) - return + // 尝试从缓存读取 + if h.cache != nil { + if cached, err := h.cache.Get(c.Request.Context(), cacheKey); err == nil && cached != "" { + var stats model.DashboardStats + if json.Unmarshal([]byte(cached), &stats) == nil { + c.JSON(http.StatusOK, model.Response{ + Code: 200, + Message: "success", + Data: stats, + }) + return + } } } @@ -231,9 +234,11 @@ func (h *DashboardHandler) GetStats(c *gin.Context) { } stats.ExpiringCerts = expiringCerts - // 写入 Redis 缓存（异步，缓存失败不影响响应） - if data, err := json.Marshal(stats); err == nil { - _ = h.cache.Set(context.Background(), cacheKey, string(data), cacheTTL) + // 写入缓存（异步，缓存失败不影响响应）// Logan.Li + if h.cache != nil { + if data, err := json.Marshal(stats); err == nil { + _ = h.cache.Set(context.Background(), cacheKey, string(data), cacheTTL) + } } c.JSON(http.StatusOK, model.Response{ @@ -327,8 +332,9 @@ func (h *DashboardHandler) GetMetrics(c *gin.Context) { AvgMem float64 } var rows []avgPoint + bucketExpr := database.HourBucketExpr(database.DriverName(h.db)) h.db.Model(&model.AgentReport{}). - Select("UNIX_TIMESTAMP(report_time) DIV 3600 as bucket_ts, ROUND(AVG(cpu_usage),2) as avg_cpu, ROUND(AVG(mem_usage),2) as avg_mem"). + Select(fmt.Sprintf("%s as bucket_ts, ROUND(AVG(cpu_usage),2) as avg_cpu, ROUND(AVG(mem_usage),2) as avg_mem", bucketExpr)). Where("report_time >= ?", since). Group("bucket_ts"). Order("bucket_ts ASC"). diff --git a/backend/controller/pipeline_executor.go b/backend/controller/pipeline_executor.go index d4605d8..3341036 100644 --- a/backend/controller/pipeline_executor.go +++ b/backend/controller/pipeline_executor.go @@ -60,8 +60,10 @@ func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.Pip entries = append(entries, entry) buf, _ := json.Marshal(entries) h.cache.Set(ctx, logKey, string(buf), 30*time.Minute) - // Pub/Sub 实时推送 - h.cache.Publish(ctx, pubCh, string(data)) + // Pub/Sub 实时推送（redka/local 无 Pub/Sub，WebSocket 走轮询）// Logan.Li + if h.cache.Capabilities().PubSub { + h.cache.Publish(ctx, pubCh, string(data)) + } } } @@ -76,7 +78,7 @@ func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.Pip } h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Updates(updates) // 推送状态变更 - if h.cache != nil { + if h.cache != nil && h.cache.Capabilities().PubSub { statusData, _ := json.Marshal(map[string]interface{}{"type": "status", "status": status, "duration": duration}) h.cache.Publish(ctx, pubCh, string(statusData)) } diff --git a/backend/main.go b/backend/main.go index 3b0c8da..cf1cea5 100644 --- a/backend/main.go +++ b/backend/main.go @@ -139,10 +139,21 @@ func main() { DataDir: "data/redka", }) if err != nil { + if cfg.Mode == "local" { + log.Fatal("local 模式缓存(redka)初始化失败", zap.Error(err)) + } log.Warn("缓存初始化失败，继续启动", zap.Error(err)) cache = nil } else { - log.Info("缓存已就绪", zap.String("driver", cfg.Redis.Driver)) + 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() } @@ -156,14 +167,18 @@ func main() { // 版本信息 API r.GET("/version", func(c *gin.Context) { - c.JSON(http.StatusOK, gin.H{ + 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) }) // 注册路由 @@ -173,13 +188,9 @@ func main() { go func() { ticker := time.NewTicker(24 * time.Hour) defer ticker.Stop() + cleanupSQL := database.CleanupAgentReportsSQL(cfg.Database.Driver) for range ticker.C { - var rows int64 - if cfg.Database.Driver == "sqlite" { - rows = db.Exec("DELETE FROM monitor_agent_reports WHERE report_time < datetime('now', '-7 days')").RowsAffected - } else { - rows = db.Exec("DELETE FROM monitor_agent_reports WHERE report_time < DATE_SUB(NOW(), INTERVAL 7 DAY)").RowsAffected - } + rows := db.Exec(cleanupSQL).RowsAffected if rows > 0 { log.Info("清理过期的 agent 上报数据", zap.Int64("rows", rows)) } diff --git a/backend/pkg/database/database.go b/backend/pkg/database/database.go index a2a0a5e..3e3cf47 100644 --- a/backend/pkg/database/database.go +++ b/backend/pkg/database/database.go @@ -2,6 +2,7 @@ package database import ( "fmt" + "strings" "time" "ezops/config" @@ -28,7 +29,18 @@ func New(cfg config.DatabaseConfig) (*gorm.DB, error) { cfg.Host, cfg.Port, cfg.User, cfg.Password, cfg.DBName) dialector = postgres.Open(dsn) case "sqlite": - dialector = sqlite.Open(cfg.DBName) + dsn := cfg.DBName + 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) } diff --git a/backend/pkg/database/dialect.go b/backend/pkg/database/dialect.go new file mode 100644 index 0000000..378d28d --- /dev/null +++ b/backend/pkg/database/dialect.go @@ -0,0 +1,43 @@ +package database + +import ( + "strings" + + "gorm.io/gorm" +) + +// DriverName 从 GORM 连接获取驱动名（sqlite / mysql / postgres）// Logan.Li +func DriverName(db *gorm.DB) string { + if db == nil || db.Dialector == nil { + return "mysql" + } + name := strings.ToLower(db.Dialector.Name()) + switch { + case strings.Contains(name, "sqlite"): + return "sqlite" + case strings.Contains(name, "postgres"): + return "postgres" + default: + return "mysql" + } +} + +// HourBucketExpr 按小时分桶表达式（跨 SQLite / MySQL）// Logan.Li +func HourBucketExpr(driver string) string { + switch driver { + case "sqlite": + return "cast(strftime('%s', report_time) as integer) / 3600" + default: + return "UNIX_TIMESTAMP(report_time) DIV 3600" + } +} + +// CleanupAgentReportsSQL 清理 7 天前 Agent 上报 // Logan.Li +func CleanupAgentReportsSQL(driver string) string { + switch driver { + case "sqlite": + return "DELETE FROM monitor_agent_reports WHERE report_time < datetime('now', '-7 days')" + default: + return "DELETE FROM monitor_agent_reports WHERE report_time < DATE_SUB(NOW(), INTERVAL 7 DAY)" + } +} \ No newline at end of file diff --git a/backend/pkg/redis/go_redis.go b/backend/pkg/redis/go_redis.go index eb7ea86..1564629 100644 --- a/backend/pkg/redis/go_redis.go +++ b/backend/pkg/redis/go_redis.go @@ -33,6 +33,10 @@ func newGoRedis(cfg Config) (Cache, error) { return &goRedisCache{client: client}, nil } +func (r *goRedisCache) Capabilities() Capabilities { + return Capabilities{SortedSet: true, PubSub: true} +} + func (r *goRedisCache) Get(ctx context.Context, key string) (string, error) { val, err := r.client.Get(ctx, key).Result() if err == goredis.Nil { diff --git a/backend/pkg/redis/redis.go b/backend/pkg/redis/redis.go index e313354..f5dc937 100644 --- a/backend/pkg/redis/redis.go +++ b/backend/pkg/redis/redis.go @@ -40,8 +40,15 @@ type SortedSetMember struct { Member string } +// Capabilities 缓存后端能力（local/redka 与外部 Redis 差异）// Logan.Li +type Capabilities struct { + SortedSet bool `json:"sorted_set"` + PubSub bool `json:"pubsub"` +} + // Cache Redis 缓存接口 // Logan.Li type Cache interface { + Capabilities() Capabilities Get(ctx context.Context, key string) (string, error) Set(ctx context.Context, key string, value string, expiration time.Duration) error Del(ctx context.Context, keys ...string) error diff --git a/backend/pkg/redis/redka.go b/backend/pkg/redis/redka.go index 2ba4d12..ea1c866 100644 --- a/backend/pkg/redis/redka.go +++ b/backend/pkg/redis/redka.go @@ -31,6 +31,10 @@ func newRedka(cfg Config) (Cache, error) { return &redkaCache{db: db}, nil } +func (r *redkaCache) Capabilities() Capabilities { + return Capabilities{SortedSet: false, PubSub: false} +} + func (r *redkaCache) Get(ctx context.Context, key string) (string, error) { // Redka v0.2.x: Str().Get() 返回 core.Value // Logan.Li val, err := r.db.Str().Get(key) @@ -115,5 +119,6 @@ func (r *redkaCache) ZRemRangeByScore(ctx context.Context, key string, min, max var errPubSubNotSupported = fmt.Errorf("Pub/Sub is not supported by the redka backend; use go-redis (driver=redis)") func (r *redkaCache) Publish(ctx context.Context, channel string, message string) error { - return errPubSubNotSupported + // local/redka 无 Pub/Sub，流水线日志走轮询 Get，此处 no-op // Logan.Li + return nil } diff --git a/deploy/kubernetes/deployment.yaml b/deploy/kubernetes/deployment.yaml index 387c3f8..202439a 100644 --- a/deploy/kubernetes/deployment.yaml +++ b/deploy/kubernetes/deployment.yaml @@ -107,7 +107,8 @@ spec: - name: uploads emptyDir: {} - name: data - emptyDir: {} + persistentVolumeClaim: + claimName: aiops-data --- apiVersion: v1 kind: Service diff --git a/deploy/kubernetes/pvc-aiops-data.yaml b/deploy/kubernetes/pvc-aiops-data.yaml new file mode 100644 index 0000000..ab8eb14 --- /dev/null +++ b/deploy/kubernetes/pvc-aiops-data.yaml @@ -0,0 +1,16 @@ +# pvc-aiops-data.yaml — local 模式 SQLite + redka 数据持久化 +# 部署: kubectl apply -f pvc-aiops-data.yaml +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: aiops-data + namespace: devops + labels: + app.kubernetes.io/name: aiops + app.kubernetes.io/component: storage +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 10Gi \ No newline at end of file diff --git a/deploy/kubernetes/quick-deploy.yaml b/deploy/kubernetes/quick-deploy.yaml index 8c6a3a4..c06619c 100644 --- a/deploy/kubernetes/quick-deploy.yaml +++ b/deploy/kubernetes/quick-deploy.yaml @@ -13,10 +13,17 @@ metadata: namespace: devops data: config.yaml: | - mode: release + mode: local port: "8080" log_level: info + database: + driver: sqlite + dbname: data/sqlite3.db + + redis: + driver: redka + jwt: secret: "aiops-jwt-secret-key-2026" expire_hour: 24 @@ -60,29 +67,17 @@ spec: value: "local" - name: TZ value: Asia/Shanghai - # 远程数据库 (MODE=release 时生效) - - name: DATABASE_DRIVER - value: "mysql" - - name: DATABASE_HOST - value: "10.10.10.75" - - name: DATABASE_PORT - value: "3306" - - name: DATABASE_USER - value: "root" - - name: DATABASE_PASSWORD - value: "aiops_root_2026" - - name: DATABASE_DBNAME - value: "aiops" - # 远程 Redis (MODE=release 时生效) - - name: REDIS_DRIVER - value: "redis" - - name: REDIS_HOST - value: "10.10.10.75" - - name: REDIS_PORT - value: "6379" - # JWT - name: JWT_SECRET value: "aiops-jwt-secret-key-2026" + # 集群模式 (MODE=release) 时取消注释: + # - name: DATABASE_DRIVER + # value: mysql + # - name: DATABASE_HOST + # value: "10.10.10.75" + # - name: REDIS_DRIVER + # value: redis + # - name: REDIS_HOST + # value: "10.10.10.75" volumeMounts: - name: config mountPath: /app/config.yaml @@ -102,7 +97,8 @@ spec: configMap: name: aiops-config - name: data - emptyDir: {} + persistentVolumeClaim: + claimName: aiops-data --- apiVersion: v1 kind: Service diff --git a/run.sh b/run.sh index 8c1043c..17b3569 100755 --- a/run.sh +++ b/run.sh @@ -451,7 +451,38 @@ save_binary_args() { printf '%s\n' "$@" > "$BINARY_ARGS_FILE" } +# 将旧版 --db-driver sqlite 参数迁移为 --mode local // Logan.Li +migrate_binary_args_file() { + [ -f "$BINARY_ARGS_FILE" ] || return 0 + if grep -qE '^--mode ' "$BINARY_ARGS_FILE" 2>/dev/null; then + return 0 + fi + if ! grep -qE '^--db-driver' "$BINARY_ARGS_FILE" 2>/dev/null; then + return 0 + fi + local port="8080" + local lines=() + while IFS= read -r line || [ -n "$line" ]; do + [ -n "$line" ] && lines+=("$line") + done < "$BINARY_ARGS_FILE" + local i=0 + while [ $i -lt ${#lines[@]} ]; do + if [ "${lines[$i]}" = "--port" ] && [ $((i + 1)) -lt ${#lines[@]} ]; then + port="${lines[$((i + 1))]}" + break + fi + if [[ "${lines[$i]}" == --port=* ]]; then + port="${lines[$i]#--port=}" + break + fi + i=$((i + 1)) + done + info "迁移启动参数: --db-driver → --mode local --port $port" + save_binary_args --mode local --port "$port" +} + load_binary_args() { + migrate_binary_args_file local -n _out=$1 _out=() if [ -f "$BINARY_ARGS_FILE" ]; then @@ -464,6 +495,21 @@ load_binary_args() { fi } +# 升级后校验 /version 驱动信息 // Logan.Li +verify_binary_version() { + local port=${1:-8080} + local body + body=$(curl -sf "http://localhost:${port}/version" 2>/dev/null) || { + warning "/version 未响应" + return 1 + } + info "版本探测: $body" + echo "$body" | grep -q '"mode":"local"' && success "mode=local 已生效" || warning "mode 非 local" + echo "$body" | grep -q '"db_driver":"sqlite"' && success "db_driver=sqlite 已生效" || true + echo "$body" | grep -q '"redis_driver":"redka"' && success "redis_driver=redka 已生效" || true + return 0 +} + binary_listen_port() { local args=("$@") local i=0 @@ -492,7 +538,7 @@ start_binary_daemon() { if wait_binary_ready "$port"; then success "二进制已启动 (PID: $(cat "$BINARY_PID_FILE"), 端口: $port)" info "日志: $BACKEND_DIR/logs/aiops.log" - info "健康检查: http://localhost:${port}/version" + verify_binary_version "$port" || true return 0 fi return 1 @@ -896,6 +942,9 @@ cmd_kubernetes() { kubectl apply -f "$KUBERNETES_MANIFEST_DIR/secret.yaml" kubectl apply -f "$KUBERNETES_MANIFEST_DIR/configmap.yaml" + info "部署 local 模式数据卷 (PVC)..." + kubectl apply -f "$KUBERNETES_MANIFEST_DIR/pvc-aiops-data.yaml" + info "部署 MySQL + Redis..." kubectl apply -f "$KUBERNETES_MANIFEST_DIR/mysql.yaml" kubectl apply -f "$KUBERNETES_MANIFEST_DIR/redis.yaml"