aiops/backend/pkg/metricagg/aggregator.go

// Package metricagg 指标分层滚动聚合。
//
// 目标：把高频原始采样（Redis ZSET）逐级降采样为多分辨率聚合行，落库供趋势查询。
//
// 层级（逐级降采样，每级只读上一级，避免每次从原始数据重算）：
//
//	raw(10s, Redis) --1min--> 1m --1h--> 1h --6h/24h--> 6h / 1d
//
// 调度（全部后台，替代原先依赖前端手动触发 POST /agent/aggregate）：
//   - fine   每 1 分钟：raw → 1m / 5m
//   - mid    每 15 分钟：1m(DB) → 1h
//   - coarse 每 1 小时：1h(DB) → 6h / 1d
//
// 幂等：依赖 monitor_metric_aggregates 的 uniq_metric_agg 唯一索引 + ON CONFLICT。 // Logan.Li
package metricagg

import (
	"context"
	"math"
	"sort"
	"strconv"
	"time"

	"ezops/model"
	ezredis "ezops/pkg/redis"

	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

// metrics 需要聚合的指标清单（与 Agent 写入 Redis 的 key 保持一致）。 // Logan.Li
var metrics = []string{"cpu", "mem", "load1"}

const (
	// rawKeyPrefix 与 controller.writeHotMetric 写入的 key 前缀保持一致。 // Logan.Li
	rawKeyPrefix = "metrics:"

	fineEvery   = time.Minute
	midEvery    = 15 * time.Minute
	coarseEvery = time.Hour

	// 各级回看窗口：只需覆盖该级「刚闭合的桶 + 边界抖动」。 // Logan.Li
	fineLookback   = 3 * time.Hour
	midLookback    = 6 * time.Hour
	coarseLookback = 8 * 24 * time.Hour
)

// fineIntervals / midIntervals / coarseIntervals 各级负责的聚合周期。 // Logan.Li
var (
	fineIntervals   = []interval{{"1m", time.Minute}, {"5m", 5 * time.Minute}}
	midIntervals    = []interval{{"1h", time.Hour}}
	coarseIntervals = []interval{{"6h", 6 * time.Hour}, {"1d", 24 * time.Hour}}
)

type interval struct {
	name string
	d    time.Duration
}

// Start 启动后台聚合调度，ctx 取消后退出。 // Logan.Li
func Start(ctx context.Context, db *gorm.DB, cache ezredis.Cache, log *zap.Logger) {
	// 无 ZSet 能力（redka/local 模式）时不做 Redis 侧聚合。 // Logan.Li
	if cache == nil || !cache.Capabilities().SortedSet {
		log.Info("指标聚合未启用：当前缓存后端不支持有序集合")
		return
	}

	go run(ctx, db, cache, log, fineEvery, "fine", func(c context.Context) {
		rollupFromRaw(c, db, cache, log, fineIntervals, fineLookback)
	})
	go run(ctx, db, cache, log, midEvery, "mid", func(c context.Context) {
		rollupFromDB(db, log, "1m", midIntervals, midLookback)
	})
	go run(ctx, db, cache, log, coarseEvery, "coarse", func(c context.Context) {
		rollupFromDB(db, log, "1h", coarseIntervals, coarseLookback)
	})
}

// run 周期性执行任务，起跑前先等一个周期，避免与启动期初始化抢资源。 // Logan.Li
func run(ctx context.Context, _ *gorm.DB, _ ezredis.Cache, log *zap.Logger, every time.Duration, name string, fn func(context.Context)) {
	ticker := time.NewTicker(every)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			start := time.Now()
			fn(ctx)
			log.Debug("指标聚合完成", zap.String("level", name), zap.Duration("cost", time.Since(start)))
		}
	}
}

// rollupFromRaw 从 Redis 原始采样聚合出细粒度桶。 // Logan.Li
func rollupFromRaw(ctx context.Context, db *gorm.DB, cache ezredis.Cache, log *zap.Logger, ivs []interval, lookback time.Duration) {
	hosts := activeHosts(db)
	if len(hosts) == 0 {
		return
	}
	now := time.Now()
	since := now.Add(-lookback)

	for _, host := range hosts {
		for _, metric := range metrics {
			key := rawKeyPrefix + metric + ":" + host
			members, err := cache.ZRangeByScore(ctx, key,
				formatUnix(since), formatUnix(now.Add(time.Minute)))
			if err != nil || len(members) == 0 {
				continue
			}
			for _, iv := range ivs {
				buckets := bucketizeMembers(members, iv.d)
				if err := upsertBuckets(db, host, metric, iv.name, buckets); err != nil {
					log.Warn("写入聚合桶失败", zap.String("host", host), zap.String("metric", metric), zap.Error(err))
				}
			}
		}
	}
}

// rollupFromDB 从上一级聚合行再聚合出粗粒度桶（逐级降采样）。 // Logan.Li
func rollupFromDB(db *gorm.DB, log *zap.Logger, sourceInterval string, ivs []interval, lookback time.Duration) {
	hosts := activeHosts(db)
	if len(hosts) == 0 {
		return
	}
	since := time.Now().Add(-lookback)

	for _, host := range hosts {
		for _, metric := range metrics {
			var rows []model.MetricsAggregate
			if err := db.Where("hostname = ? AND metric = ? AND `interval` = ? AND bucket_time >= ?",
				host, metric, sourceInterval, since).
				Order("bucket_time ASC").Find(&rows).Error; err != nil || len(rows) == 0 {
				continue
			}
			samples := make([]weightedSample, 0, len(rows))
			for _, r := range rows {
				samples = append(samples, weightedSample{ts: r.BucketTime, avg: r.Avg, count: r.Count})
			}
			for _, iv := range ivs {
				buckets := bucketizeSamples(samples, iv.d)
				if err := upsertBuckets(db, host, metric, iv.name, buckets); err != nil {
					log.Warn("写入降采样桶失败", zap.String("host", host), zap.String("metric", metric), zap.Error(err))
				}
			}
		}
	}
}

// activeHosts 只聚合近期在线的 Agent 主机，避免对已下线主机做无用扫描。 // Logan.Li
func activeHosts(db *gorm.DB) []string {
	var hosts []string
	cutoff := time.Now().Add(-15 * time.Minute)
	db.Model(&model.Server{}).
		Where("agent_status = ? AND last_check_at >= ?", "online", cutoff).
		Distinct().Pluck("hostname", &hosts)
	return hosts
}

type bucketStats struct {
	avg, maxV, minV, p95 float64
	count                int
}

type weightedSample struct {
	ts    time.Time
	avg   float64
	count int
}

// bucketizeMembers 把原始采样按周期分桶。 // Logan.Li
func bucketizeMembers(members []ezredis.SortedSetMember, d time.Duration) map[time.Time][]float64 {
	sec := int64(d.Seconds())
	out := make(map[time.Time][]float64)
	for _, m := range members {
		ts := int64(m.Score)
		start := (ts / sec) * sec
		out[time.Unix(start, 0).UTC()] = append(out[time.Unix(start, 0).UTC()], parseValue(m.Member))
	}
	return out
}

// bucketizeSamples 把上一级聚合行按周期再聚合（用 count 加权，避免小样本被放大）。 // Logan.Li
func bucketizeSamples(samples []weightedSample, d time.Duration) map[time.Time][]float64 {
	sec := int64(d.Seconds())
	out := make(map[time.Time][]float64)
	for _, s := range samples {
		ts := s.ts.Unix()
		start := (ts / sec) * sec
		key := time.Unix(start, 0).UTC()
		// 用 avg 重复 count 次近似还原样本分布，保证 max/min/p95 有意义。 // Logan.Li
		n := s.count
		if n < 1 {
			n = 1
		}
		if n > 1000 {
			n = 1000
		}
		for i := 0; i < n; i++ {
			out[key] = append(out[key], s.avg)
		}
	}
	return out
}

// upsertBuckets 幂等写入聚合行（依赖 uniq_metric_agg 唯一索引）。 // Logan.Li
func upsertBuckets(db *gorm.DB, host, metric, intervalName string, buckets map[time.Time][]float64) error {
	for bucketTime, values := range buckets {
		if len(values) == 0 {
			continue
		}
		st := calcStats(values)
		row := model.MetricsAggregate{
			Hostname:   host,
			Metric:     metric,
			Interval:   intervalName,
			Avg:        st.avg,
			MaxV:       st.maxV,
			MinV:       st.minV,
			P95:        st.p95,
			Count:      st.count,
			BucketTime: bucketTime,
		}
		if err := 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(&row).Error; err != nil {
			return err
		}
	}
	return nil
}

// calcStats 计算 avg / max / min / p95，结果保留两位小数。 // Logan.Li
func calcStats(values []float64) bucketStats {
	sorted := make([]float64, len(values))
	copy(sorted, values)
	sort.Float64s(sorted)

	var sum float64
	for _, v := range sorted {
		sum += v
	}
	avg := sum / float64(len(sorted))
	p95 := sorted[int(math.Ceil(float64(len(sorted))*0.95))-1]

	return bucketStats{
		avg:   round2(avg),
		maxV:  round2(sorted[len(sorted)-1]),
		minV:  round2(sorted[0]),
		p95:   round2(p95),
		count: len(sorted),
	}
}

func round2(v float64) float64 { return math.Round(v*100) / 100 }

// parseValue 从 "45.2300:1234567890123456789" 中取出数值部分。 // Logan.Li
func parseValue(member string) float64 {
	for i := 0; i < len(member); i++ {
		if member[i] == ':' {
			return parseFloat(member[:i])
		}
	}
	return parseFloat(member)
}

func parseFloat(s string) float64 {
	var v float64
	var frac, div float64 = 0, 1
	neg := false
	i := 0
	if i < len(s) && (s[i] == '-' || s[i] == '+') {
		neg = s[i] == '-'
		i++
	}
	for ; i < len(s); i++ {
		c := s[i]
		if c == '.' {
			i++
			break
		}
		if c < '0' || c > '9' {
			break
		}
		v = v*10 + float64(c-'0')
	}
	for ; i < len(s); i++ {
		c := s[i]
		if c < '0' || c > '9' {
			break
		}
		frac = frac*10 + float64(c-'0')
		div *= 10
	}
	v += frac / div
	if neg {
		v = -v
	}
	return v
}

func formatUnix(t time.Time) string {
	// Redis ZSET score 为 Unix 秒；用整数文本作为 ZRANGEBYSCORE 的边界。 // Logan.Li
	return strconv.FormatInt(t.Unix(), 10)
}