opsx
c090ba0564
全命名空间 6h/24h 聚合数据量过大会拖垮 Loki:前端时间范围旁新增应用范围选择器 (labelValues 候选),>1h 未选应用禁用开始分析并提示;后端同样强制 400(闸门在服务端); 信号采集选择器按 app 收窄后 histogram/stats 从全量变为单应用流,秒级返回。
285 lines
9.5 KiB
Go
285 lines
9.5 KiB
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, app string) (*AISignalBundle, error) {
|
||
if src == nil {
|
||
return nil, fmt.Errorf("数据源不存在")
|
||
}
|
||
if !to.After(from) {
|
||
return nil, fmt.Errorf("时间范围无效")
|
||
}
|
||
// 复用查询授权：AI 能看到的数据范围必须与当前用户一致，不多不少。
|
||
// 长窗口由上层强制指定 app(缩小聚合范围);15m/1h 允许全命名空间
|
||
baseSelector := `{namespace=~".+"}`
|
||
if app != "" {
|
||
baseSelector = `{namespace=~".+", app="` + app + `"}`
|
||
}
|
||
selector, err := AuthorizedLogQuery(src, userID, admin, baseSelector)
|
||
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()
|
||
}
|