rlwj-lab/aiops
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Redis热数据+K线聚合+Dashboard MySQL指标修复

Browse Source
This commit is contained in:
attacker committed 2026-06-18 08:03:49 +00:00
1 parent b8b1a3d5f6
commit a843ab7b00
9 files changed
+311 -12
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/internal/api/handlers/agent.go
+200 -3
View File
@@ -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
}
backend/internal/api/handlers/dashboard.go
+4 -4
View File
@@ -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)
backend/internal/api/routes/routes.go
+9 -5
View File
@@ -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)
backend/internal/models/metrics_kline.go
+38
View File
@@ -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
}
backend/pkg/database/database.go
+1
View File
@@ -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
backend/pkg/redis/go_redis.go
+29
View File
@@ -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()
}
backend/pkg/redis/redis.go
+11
View File
@@ -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
backend/pkg/redis/redka.go
+15
View File
@@ -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
}