diff --git a/backend/internal/api/handlers/agent.go b/backend/internal/api/handlers/agent.go index e98a86d..c6df9c2 100644 --- a/backend/internal/api/handlers/agent.go +++ b/backend/internal/api/handlers/agent.go @@ -3,22 +3,28 @@ package handlers import ( "encoding/json" "fmt" + "math" "net/http" + "strconv" + "strings" "time" "ezops/internal/models" + ezredis "ezops/pkg/redis" "github.com/gin-gonic/gin" "gorm.io/gorm" + "gorm.io/gorm/clause" ) // AgentHandler 处理 Agent 上报数据 // Logan.Li type AgentHandler struct { - db *gorm.DB + db *gorm.DB + cache ezredis.Cache // 可选 Redis 缓存（用于热数据存储） } -func NewAgentHandler(db *gorm.DB) *AgentHandler { - return &AgentHandler{db: db} +func NewAgentHandler(db *gorm.DB, cache ezredis.Cache) *AgentHandler { + return &AgentHandler{db: db, cache: cache} } // ReceiveReport 接收 Agent 上报的指标数据 @@ -63,10 +69,44 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { return } + // 写入 Redis 有序集合（热数据，保留7天） + if h.cache != nil { + timestamp := float64(reportTime.Unix()) + ttl := 7 * 24 * time.Hour // 7天 + + // 使用 reportTime.UnixNano() 确保 member 唯一（避免相同值碰撞） + 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) + } + // 保存成功 c.JSON(http.StatusOK, models.Response{Code: 200, Message: "ok"}) } +// writeHotMetric 将单个指标写入 Redis 有序集合并维护 TTL 和过期数据清理 +func (h *AgentHandler) writeHotMetric(c *gin.Context, key string, value float64, score float64, ttl time.Duration, reportTime time.Time) { + ctx := c.Request.Context() + + // member 格式: "value:nano_timestamp" 保证唯一性 + member := fmt.Sprintf("%.4f:%d", value, reportTime.UnixNano()) + + if err := h.cache.ZAdd(ctx, key, ezredis.SortedSetMember{ + Score: score, + Member: member, + }); err != nil { + // 静默失败，不影响主流程 + return + } + + // 设置 key 过期时间 + _ = h.cache.Expire(ctx, key, ttl) + + // 清理 7 天前的旧数据 + cutoff := float64(time.Now().Add(-ttl).Unix()) + _ = h.cache.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%.0f", cutoff)) +} + // GetLatestReports 获取各主机最新上报 // GET /api/v1/agent/latest func (h *AgentHandler) GetLatestReports(c *gin.Context) { @@ -108,3 +148,160 @@ func (h *AgentHandler) GetHostHistory(c *gin.Context) { Data: reports, }) } + +// AggregateMetrics 聚合 Redis 热数据到 MySQL klines +// POST /api/v1/agent/aggregate +func (h *AgentHandler) AggregateMetrics(c *gin.Context) { + var req models.AggregateKlineReq + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求格式错误: " + err.Error()}) + return + } + + if h.cache == nil { + c.JSON(http.StatusServiceUnavailable, models.Response{Code: 503, Message: "Redis 缓存不可用"}) + return + } + + // 支持的聚合间隔及对应的回溯时长 + intervals := []struct { + name string + duration time.Duration + }{ + {"1m", 1 * time.Minute}, + {"3m", 3 * time.Minute}, + {"1h", 1 * time.Hour}, + {"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 + + 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 { + 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) + } + + // 为每个桶计算 OHLC + Avg + for bucketTs, values := range buckets { + if len(values) == 0 { + continue + } + open := values[0] + closeV := values[len(values)-1] + high := values[0] + low := values[0] + sum := 0.0 + + for _, v := range values { + sum += v + if v > high { + high = v + } + if v < low { + low = v + } + } + avg := sum / float64(len(values)) + + kline := models.MetricsKline{ + Hostname: req.Hostname, + Metric: req.Metric, + Interval: iv.name, + Open: math.Round(open*100) / 100, + High: math.Round(high*100) / 100, + Low: math.Round(low*100) / 100, + Close: math.Round(closeV*100) / 100, + Avg: math.Round(avg*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{ + "open", "high", "low", "close", "avg", "count", + }), + }).Create(&kline) + + if result.Error == nil { + totalUpserted++ + } + } + } + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: fmt.Sprintf("聚合完成，写入/更新 %d 条记录", totalUpserted), + Data: gin.H{"upserted": totalUpserted}, + }) +} + +// GetKline 查询 K 线数据 +// GET /api/v1/agent/kline?hostname=&metric=&interval=&from=&to= +func (h *AgentHandler) GetKline(c *gin.Context) { + var req models.KlineQueryReq + if err := c.ShouldBindQuery(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) + return + } + + from, err := time.Parse(time.RFC3339, req.From) + if err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "from 格式错误，需 RFC3339: " + err.Error()}) + return + } + to, err := time.Parse(time.RFC3339, req.To) + if err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "to 格式错误，需 RFC3339: " + err.Error()}) + return + } + + var klines []models.MetricsKline + h.db.Where("hostname = ? AND metric = ? AND interval = ? AND bucket_time >= ? AND bucket_time <= ?", + req.Hostname, req.Metric, req.Interval, from, to). + Order("bucket_time ASC"). + Find(&klines) + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: "success", + Data: klines, + }) +} + +// parseMemberValue 从 member 字符串 "45.2300:1234567890123456789" 中提取浮点值 +func parseMemberValue(member string) float64 { + idx := strings.Index(member, ":") + if idx < 0 { + v, _ := strconv.ParseFloat(member, 64) + return v + } + v, _ := strconv.ParseFloat(member[:idx], 64) + return v +} diff --git a/backend/internal/api/handlers/dashboard.go b/backend/internal/api/handlers/dashboard.go index c9d73e1..2b1863c 100644 --- a/backend/internal/api/handlers/dashboard.go +++ b/backend/internal/api/handlers/dashboard.go @@ -294,17 +294,17 @@ func (h *DashboardHandler) GetMetrics(c *gin.Context) { } } - // 从 agent_reports 表查最近 N 小时的真实数据 // Logan.Li + // 从 agent_reports 表查最近 N 小时的真实数据（MySQL 聚合） since := time.Now().Add(-time.Duration(hours) * time.Hour) type avgPoint struct { - Bucket string // strftime 返回字符串 "2006-01-02 15:00"，不能直接扫入 time.Time（会得零值） + Bucket string AvgCPU float64 AvgMem float64 } var rows []avgPoint h.db.Model(&models.AgentReport{}). - Select("strftime('%Y-%m-%d %H:00', created_at) as bucket, AVG(cpu_usage) as avg_cpu, AVG(mem_usage) as avg_mem"). - Where("created_at >= ?", since). + Select("DATE_FORMAT(report_time, '%Y-%m-%d %H:00') as bucket, ROUND(AVG(cpu_usage),2) as avg_cpu, ROUND(AVG(mem_usage),2) as avg_mem"). + Where("report_time >= ?", since). Group("bucket"). Order("bucket ASC"). Scan(&rows) diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index 0a6def1..af16739 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -4,6 +4,7 @@ import ( "ezops/internal/api/handlers" "ezops/internal/api/middleware" "ezops/internal/config" + ezredis "ezops/pkg/redis" "fmt" "github.com/gin-gonic/gin" @@ -63,6 +64,9 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c api.POST("/auth/login", authHandler.Login) api.POST("/auth/register", authHandler.Register) + // Agent 安装处理器（在 servers 和 agent 路由中共享） + agentInstallHandler := handlers.NewAgentInstallHandler(db) + // 需要认证的路由 authorized := api.Group("") authorized.Use(middleware.JWTAuth(&cfg.JWT)) @@ -139,12 +143,8 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c servers.POST("/batch", serverHandler.BatchCreateServers) // Agent 远程安装 // Logan.Li - agentInstallHandler := handlers.NewAgentInstallHandler(db) servers.POST("/:id/install-agent", agentInstallHandler.InstallAgent) servers.POST("/install-agent", agentInstallHandler.BatchInstallAgent) - - // Agent 远程命令执行 - agentGroup.POST("/exec", agentInstallHandler.RemoteExec) } // 云账号路由 @@ -413,19 +413,23 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c } // Agent 上报路由（无需 JWT） - agentHandler := handlers.NewAgentHandler(db) + redisCache, _ := cache.(ezredis.Cache) + agentHandler := handlers.NewAgentHandler(db, redisCache) taskHandler := handlers.NewAgentTaskHandler(db) agentGroup := api.Group("/agent") { agentGroup.POST("/report", agentHandler.ReceiveReport) agentGroup.GET("/tasks/poll", taskHandler.PollTasks) agentGroup.POST("/tasks/:task_id/result", taskHandler.SubmitResult) + agentGroup.POST("/aggregate", agentHandler.AggregateMetrics) + agentGroup.POST("/exec", agentInstallHandler.RemoteExec) } // Agent 数据查询 + 任务管理（需要 JWT） agentQuery := authorized.Group("/agent") { agentQuery.GET("/latest", agentHandler.GetLatestReports) agentQuery.GET("/history/:hostname", agentHandler.GetHostHistory) + agentQuery.GET("/kline", agentHandler.GetKline) // 远程任务 agentQuery.POST("/tasks", taskHandler.CreateTask) diff --git a/backend/internal/models/metrics_kline.go b/backend/internal/models/metrics_kline.go new file mode 100644 index 0000000..0fb7270 --- /dev/null +++ b/backend/internal/models/metrics_kline.go @@ -0,0 +1,38 @@ +package models + +import "time" + +// MetricsKline 指标K线聚合数据（由 Redis 热数据聚合到 MySQL）// Logan.Li +type MetricsKline struct { + ID uint `gorm:"primarykey" json:"id"` + Hostname string `gorm:"size:255;index:idx_kline;not null" json:"hostname"` + Metric string `gorm:"size:32;index:idx_kline;not null" json:"metric"` // cpu / mem / load1 + Interval string `gorm:"size:8;index:idx_kline;not null" json:"interval"` // 1m / 3m / 1h / 6h / 1d + Open float64 `gorm:"type:decimal(10,2)" json:"open"` + High float64 `gorm:"type:decimal(10,2)" json:"high"` + Low float64 `gorm:"type:decimal(10,2)" json:"low"` + Close float64 `gorm:"type:decimal(10,2)" json:"close"` + Avg float64 `gorm:"type:decimal(10,2)" json:"avg"` + Count int `json:"count"` + BucketTime time.Time `gorm:"index:idx_kline;not null" json:"bucket_time"` + CreatedAt time.Time `json:"created_at"` +} + +func (MetricsKline) TableName() string { + return "metrics_klines" +} + +// AggregateKlineReq 聚合请求体 +type AggregateKlineReq struct { + Hostname string `json:"hostname" binding:"required"` + Metric string `json:"metric" binding:"required"` // cpu / mem / load1 +} + +// KlineQueryReq K线查询参数（Query 参数） +type KlineQueryReq struct { + Hostname string `form:"hostname" binding:"required"` + Metric string `form:"metric" binding:"required"` + Interval string `form:"interval" binding:"required"` + From string `form:"from" binding:"required"` // RFC3339 + To string `form:"to" binding:"required"` // RFC3339 +} diff --git a/backend/pkg/database/database.go b/backend/pkg/database/database.go index e0bb855..2770cd5 100644 --- a/backend/pkg/database/database.go +++ b/backend/pkg/database/database.go @@ -106,6 +106,7 @@ func AutoMigrate(db *gorm.DB) error { &models.TerminalSession{}, // 终端会话录制（终端审计） // Logan.Li &models.LLMConfig{}, // 大语言模型配置（AI 智能助手） // Logan.Li &models.ChatHistory{}, // 对话历史记录（按用户隔离） // Logan.Li + &models.MetricsKline{}, // 指标K线聚合（Redis 热数据 → MySQL） // Logan.Li } var lastErr error diff --git a/backend/pkg/redis/go_redis.go b/backend/pkg/redis/go_redis.go index 58c18fb..6834e4e 100644 --- a/backend/pkg/redis/go_redis.go +++ b/backend/pkg/redis/go_redis.go @@ -64,3 +64,32 @@ func (r *goRedisCache) Ping(ctx context.Context) error { func (r *goRedisCache) Close() error { return r.client.Close() } + +func (r *goRedisCache) ZAdd(ctx context.Context, key string, member SortedSetMember) error { + return r.client.ZAdd(ctx, key, goredis.Z{ + Score: member.Score, + Member: member.Member, + }).Err() +} + +func (r *goRedisCache) ZRangeByScore(ctx context.Context, key string, min, max string) ([]SortedSetMember, error) { + zs, err := r.client.ZRangeByScoreWithScores(ctx, key, &goredis.ZRangeBy{ + Min: min, + Max: max, + }).Result() + if err != nil { + return nil, err + } + members := make([]SortedSetMember, len(zs)) + for i, z := range zs { + members[i] = SortedSetMember{ + Score: z.Score, + Member: z.Member.(string), + } + } + return members, nil +} + +func (r *goRedisCache) ZRemRangeByScore(ctx context.Context, key string, min, max string) error { + return r.client.ZRemRangeByScore(ctx, key, min, max).Err() +} diff --git a/backend/pkg/redis/redis.go b/backend/pkg/redis/redis.go index de697f3..e947b21 100644 --- a/backend/pkg/redis/redis.go +++ b/backend/pkg/redis/redis.go @@ -34,6 +34,12 @@ type Config struct { DataDir string // redka 数据目录 } +// SortedSetMember 有序集合成员 // Logan.Li +type SortedSetMember struct { + Score float64 + Member string +} + // Cache Redis 缓存接口 // Logan.Li type Cache interface { Get(ctx context.Context, key string) (string, error) @@ -43,6 +49,11 @@ type Cache interface { Expire(ctx context.Context, key string, expiration time.Duration) error Ping(ctx context.Context) error Close() error + + // 有序集合操作（热数据指标存储，仅 go-redis 后端完整支持） + ZAdd(ctx context.Context, key string, member SortedSetMember) error + ZRangeByScore(ctx context.Context, key string, min, max string) ([]SortedSetMember, error) + ZRemRangeByScore(ctx context.Context, key string, min, max string) error } // New 创建缓存实例（根据驱动自动选择后端）// Logan.Li diff --git a/backend/pkg/redis/redka.go b/backend/pkg/redis/redka.go index 33f7a0a..88f8ddc 100644 --- a/backend/pkg/redis/redka.go +++ b/backend/pkg/redis/redka.go @@ -5,6 +5,7 @@ package redis import ( "context" + "fmt" "time" "github.com/nalgeon/redka" @@ -96,3 +97,17 @@ func (r *redkaCache) Ping(ctx context.Context) error { func (r *redkaCache) Close() error { return r.db.Close() } + +var errSortedSetNotSupported = fmt.Errorf("sorted set operations are not supported by the redka backend; use go-redis (driver=redis) instead") + +func (r *redkaCache) ZAdd(ctx context.Context, key string, member SortedSetMember) error { + return errSortedSetNotSupported +} + +func (r *redkaCache) ZRangeByScore(ctx context.Context, key string, min, max string) ([]SortedSetMember, error) { + return nil, errSortedSetNotSupported +} + +func (r *redkaCache) ZRemRangeByScore(ctx context.Context, key string, min, max string) error { + return errSortedSetNotSupported +} diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index 3e55f5a..d9844f7 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -491,6 +491,10 @@ export const agentApi = { listTasks: (params?: Record) => http.get('/agent/tasks', params), // 直连 SSH 执行命令（即时返回） execCommand: (data: { hostname: string; command: string; timeout?: number }) => http.post('/agent/exec', data), + // K线聚合查询 + getKline: (params: { hostname: string; metric: string; interval: string; from?: string; to?: string }) => http.get('/agent/kline', params), + // 触发聚合 + aggregate: (data: { hostname: string; metric: string }) => http.post('/agent/aggregate', data), listRules: () => http.get('/agent/rules'), createRule: (data: any) => http.post('/agent/rules', data), updateRule: (id: number, data: any) => http.put(`/agent/rules/${id}`, data),