LokiStack/backend/services/ai_signal.go

package services

import (
	"context"
	"fmt"
	"os"
	"sort"
	"strings"
	"time"

	"lokistack/models"
	"lokistack/repository"
)

// AI 分析的"信号包"：把 GB 级日志压成 KB 级结构化数据。
//
// 成本优先原则（P0 的硬约束）：
//   - 体积/趋势 → index/volume_range（实测 24h 仅 54ms，不读日志内容）
//   - 条数/流数 → index/stats，只允许在 ≤2 小时窗口上用
//     （全天 index/stats 实测 20 秒，绝不能进 AI 路径）
//   - 应用排行 → 平台按小时预聚合的快照（毫秒级）
//   - 告警 → 平台库
//
// 也就是说：**P0 完全不扫日志内容**，一次分析对 Loki 的压力接近 0。
type AISignalBundle struct {
	SourceID     uint   `json:"sourceId"`
	SourceName   string `json:"sourceName"`
	Selector     string `json:"selector"`
	From         string `json:"from"`
	To           string `json:"to"`
	RangeMinutes int    `json:"rangeMinutes"`

	TotalBytes   int64 `json:"totalBytes"`
	TotalEntries int64 `json:"totalEntries"`
	// EntriesExact 条数是否为精确值（false = 范围过大跳过了 index/stats）
	EntriesExact bool `json:"entriesExact"`

	Streams int64 `json:"streams"`
	Chunks  int64 `json:"chunks"`

	Trend []AITrendPoint `json:"trend"`
	Apps  []AISignalApp  `json:"apps"`

	Baseline *AIBaseline `json:"baseline,omitempty"`
	Alerts   []AIAlert   `json:"alerts"`

	// SnapshotHours 平台按小时预聚合已留存的小时数（影响排行的可信度）
	SnapshotHours int `json:"snapshotHours"`
	// Warnings 采集过程中的降级说明（如"范围过大，条数为估算"）
	Warnings []string `json:"warnings"`
}

// aiDisplayZone 结论中展示时间用的时区。
//
// 为什么不直接用 time.Local：静态编译的容器镜像通常没有 /usr/share/zoneinfo，
// TZ 也没设，time.Local 就是 UTC —— 结果是模型在结论里写"凌晨 6 点日志下跌"，
// 而运维当时是下午 2 点，直接被当成数据错误。
// 默认东八区（本地运维环境），需要别的时区可用 LOKISTACK_DISPLAY_TZ=UTC 覆盖。
func aiDisplayZone() *time.Location {
	if strings.EqualFold(strings.TrimSpace(os.Getenv("LOKISTACK_DISPLAY_TZ")), "UTC") {
		return time.UTC
	}
	return time.FixedZone("CST", 8*3600)
}

type AITrendPoint struct {
	Time  string `json:"time"`
	Bytes int64  `json:"bytes"`
}

type AISignalApp struct {
	App       string `json:"app"`
	Namespace string `json:"namespace"`
	Entries   int64  `json:"entries"`
	Bytes     int64  `json:"bytes"`
	Share     string `json:"share"`
}

type AIBaseline struct {
	// PreviousBytes 上一周期同时长的体积（等长窗口对比）
	PreviousBytes  int64   `json:"previousBytes"`
	CurrentBytes   int64   `json:"currentBytes"`
	DeltaPercent   float64 `json:"deltaPercent"`
	ComparisonNote string  `json:"comparisonNote"`
}

type AIAlert struct {
	RuleName  string `json:"ruleName"`
	Status    string `json:"status"`
	Count     int    `json:"count"`
	CreatedAt string `json:"createdAt"`
}

// CollectAISignals 采集一次分析所需的全部信号。
func CollectAISignals(ctx context.Context, src *models.LogSource, userID uint, admin bool, from, to time.Time) (*AISignalBundle, error) {
	if src == nil {
		return nil, fmt.Errorf("数据源不存在")
	}
	if !to.After(from) {
		return nil, fmt.Errorf("时间范围无效")
	}
	// 复用查询授权：AI 能看到的数据范围必须与当前用户一致，不多不少
	selector, err := AuthorizedLogQuery(src, userID, admin, `{namespace=~".+"}`)
	if err != nil {
		return nil, err
	}

	bundle := &AISignalBundle{
		SourceID:   src.ID,
		SourceName: src.Name,
		Selector:   selector,
		// 时间统一按本地时区展示：Loki 返回的是 UTC，
		// 直接给模型 UTC 会让结论里出现"凌晨 6 点"这种和用户作息对不上的时间。
		From:         from.In(aiDisplayZone()).Format("2006-01-02 15:04"),
		To:           to.In(aiDisplayZone()).Format("2006-01-02 15:04"),
		RangeMinutes: int(to.Sub(from).Minutes()),
		Alerts:       []AIAlert{},
		Trend:        []AITrendPoint{},
		Apps:         []AISignalApp{},
	}

	// ── 体积 + 趋势：volume_range（只读索引元数据）──
	buckets, err := LokiHistogram(ctx, src, selector, from.UnixNano(), to.UnixNano())
	if err != nil {
		return nil, err
	}
	for _, b := range buckets {
		var bytes int64
		switch v := b["count"].(type) {
		case int64:
			bytes = v
		case float64:
			bytes = int64(v)
		}
		ts, _ := b["timestamp"].(string)
		if parsed, perr := time.Parse(time.RFC3339, ts); perr == nil {
			ts = parsed.In(aiDisplayZone()).Format("15:04")
		}
		bundle.TotalBytes += bytes
		bundle.Trend = append(bundle.Trend, AITrendPoint{Time: ts, Bytes: bytes})
	}
	// 趋势只保留尾部若干桶，避免 prompt 过长
	if len(bundle.Trend) > 24 {
		bundle.Trend = bundle.Trend[len(bundle.Trend)-24:]
	}

	// ── 条数：仅短窗口走 index/stats（长窗口太慢，直接跳过并标注）──
	if bundle.RangeMinutes <= 120 {
		if st, statErr := LokiStatsCached(ctx, src, selector, from.UnixNano(), to.UnixNano()); statErr == nil {
			bundle.TotalEntries = st.Entries
			bundle.Streams = st.Streams
			bundle.Chunks = st.Chunks
			bundle.EntriesExact = true
		} else {
			bundle.Warnings = append(bundle.Warnings, "条数统计失败："+truncateStr(statErr.Error(), 120))
		}
	} else {
		bundle.Warnings = append(bundle.Warnings, "时间范围超过 2 小时，条数为估算（跳过 index/stats：全天统计实测需 20 秒）")
	}

	// ── 应用排行：平台按小时预聚合快照 ──
	if rows := GetAppRanking(src.ID, from, to, 10, "entries"); len(rows) > 0 {
		var total int64
		for _, r := range rows {
			total += r.Entries
		}
		for _, r := range rows {
			share := ""
			if total > 0 {
				share = fmt.Sprintf("%.1f%%", float64(r.Entries)/float64(total)*100)
			}
			bundle.Apps = append(bundle.Apps, AISignalApp{
				App: r.App, Namespace: r.Namespace, Entries: r.Entries, Bytes: r.Bytes, Share: share,
			})
		}
	}
	hours, _ := AppVolumeCoverageInfo(src.ID)
	bundle.SnapshotHours = hours
	if len(bundle.Apps) == 0 {
		bundle.Warnings = append(bundle.Warnings, "应用排行暂无数据：平台按小时预聚合还在积累（当前留存 0 小时）")
	}

	// ── 基线对比：与上一个等长窗口比 ──
	prevFrom := from.Add(-to.Sub(from))
	prevTo := from
	if prevBuckets, perr := LokiHistogram(ctx, src, selector, prevFrom.UnixNano(), prevTo.UnixNano()); perr == nil {
		var prevBytes int64
		for _, b := range prevBuckets {
			switch v := b["count"].(type) {
			case int64:
				prevBytes += v
			case float64:
				prevBytes += int64(v)
			}
		}
		if prevBytes > 0 || bundle.TotalBytes > 0 {
			delta := 0.0
			if prevBytes > 0 {
				delta = (float64(bundle.TotalBytes) - float64(prevBytes)) / float64(prevBytes) * 100
			}
			bundle.Baseline = &AIBaseline{
				PreviousBytes:  prevBytes,
				CurrentBytes:   bundle.TotalBytes,
				DeltaPercent:   delta,
				ComparisonNote: fmt.Sprintf("与上一个 %d 分钟等长窗口对比", bundle.RangeMinutes),
			}
		}
	}

	// ── 告警：窗口内的投递记录 ──
	db := repository.GetDB()
	if db != nil {
		db.RLock()
		for _, d := range db.AlertDeliveries {
			if d.CreatedAt.Before(from) || d.CreatedAt.After(to) {
				continue
			}
			bundle.Alerts = append(bundle.Alerts, AIAlert{
				RuleName: d.RuleName, Status: d.Status, Count: d.Count,
				CreatedAt: d.CreatedAt.In(aiDisplayZone()).Format("2006-01-02 15:04"),
			})
		}
		db.RUnlock()
	}
	sort.Slice(bundle.Alerts, func(i, j int) bool { return bundle.Alerts[i].CreatedAt > bundle.Alerts[j].CreatedAt })
	if len(bundle.Alerts) > 10 {
		bundle.Alerts = bundle.Alerts[:10]
	}
	return bundle, nil
}

// SignalsAsText 把信号包渲染成给模型看的紧凑文本（不是 JSON，省 token）。
func SignalsAsText(b *AISignalBundle) string {
	var sb strings.Builder
	fmt.Fprintf(&sb, "数据源：%s\n", b.SourceName)
	fmt.Fprintf(&sb, "时间范围：%s ~ %s（%d 分钟）\n", b.From, b.To, b.RangeMinutes)
	fmt.Fprintf(&sb, "日志体积：%.2f GB\n", float64(b.TotalBytes)/1024/1024/1024)
	if b.EntriesExact {
		fmt.Fprintf(&sb, "日志条数：%d\n", b.TotalEntries)
	} else {
		sb.WriteString("日志条数：（范围过大未精确统计）\n")
	}
	if b.Streams > 0 || b.Chunks > 0 {
		fmt.Fprintf(&sb, "流数/块数：%d / %d\n", b.Streams, b.Chunks)
	}
	if b.Baseline != nil {
		fmt.Fprintf(&sb, "基线对比：上一等长窗口 %.2f GB → 当前 %.2f GB，变化 %+.1f%%（%s）\n",
			float64(b.Baseline.PreviousBytes)/1024/1024/1024,
			float64(b.Baseline.CurrentBytes)/1024/1024/1024,
			b.Baseline.DeltaPercent, b.Baseline.ComparisonNote)
	}
	if len(b.Apps) > 0 {
		sb.WriteString("\n应用日志量排行（来自平台小时级预聚合）：\n")
		for i, a := range b.Apps {
			fmt.Fprintf(&sb, "  %d. %s/%s  条数=%d 占比=%s\n", i+1, a.Namespace, a.App, a.Entries, a.Share)
		}
	}
	if len(b.Trend) > 0 {
		sb.WriteString("\n体积趋势（最近分桶）：\n")
		for _, t := range b.Trend {
			fmt.Fprintf(&sb, "  %s  %.1f MB\n", t.Time, float64(t.Bytes)/1024/1024)
		}
	}
	if len(b.Alerts) > 0 {
		sb.WriteString("\n窗口内告警：\n")
		for _, a := range b.Alerts {
			fmt.Fprintf(&sb, "  %s  %s  命中=%d  状态=%s\n", a.CreatedAt, a.RuleName, a.Count, a.Status)
		}
	} else {
		sb.WriteString("\n窗口内告警：无\n")
	}
	if len(b.Warnings) > 0 {
		sb.WriteString("\n数据限制说明：\n")
		for _, w := range b.Warnings {
			fmt.Fprintf(&sb, "  - %s\n", w)
		}
	}
	return sb.String()
}