优化 local 模式兼容性与部署可靠性

- SQL 方言层：agent updated_at、dashboard 小时分桶、清理任务跨 SQLite/MySQL
- Cache Capabilities：redka 跳过 ZSet/PubSub，Agent 聚合改 DB 直读
- local 模式 redka 初始化失败 Fatal；/version 返回 features
- SQLite WAL + 连接池限制；K8s data 改 PVC 持久化
- run.sh：旧参数迁移、升级后 /version 校验

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）

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").

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))
 		}

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))
 			}

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)
 	}

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)"
+	}
+}

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 {

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

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
 }