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() }