opsx
c090ba0564
全命名空间 6h/24h 聚合数据量过大会拖垮 Loki:前端时间范围旁新增应用范围选择器 (labelValues 候选),>1h 未选应用禁用开始分析并提示;后端同样强制 400(闸门在服务端); 信号采集选择器按 app 收窄后 histogram/stats 从全量变为单应用流,秒级返回。
244 lines
8.0 KiB
Go
244 lines
8.0 KiB
Go
package services
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"lokistack/models"
|
||
"lokistack/repository"
|
||
)
|
||
|
||
// AIAnalysisResult 结构化结论。
|
||
//
|
||
// 强制模型输出这个结构而不是自由文本，原因：
|
||
// 1. 前端能分区渲染（结论/证据/建议/不确定），可读性远好于一段散文；
|
||
// 2. 每条结论都要挂 evidence，逼模型把话落在数据上，降低编造；
|
||
// 3. unknowns 单独一栏，明确"不知道什么"比编一个答案有价值得多。
|
||
type AIAnalysisResult struct {
|
||
Summary string `json:"summary"`
|
||
Findings []AIFinding `json:"findings"`
|
||
Actions []string `json:"actions"`
|
||
Unknowns []string `json:"unknowns"`
|
||
RawText string `json:"-"`
|
||
ParseErr string `json:"parseError,omitempty"`
|
||
}
|
||
|
||
type AIFinding struct {
|
||
Title string `json:"title"`
|
||
Evidence string `json:"evidence"`
|
||
Severity string `json:"severity"` // high | medium | low
|
||
}
|
||
|
||
// AIAnalysisOutcome 一次分析的完整返回（含成本信息）。
|
||
type AIAnalysisOutcome struct {
|
||
Result AIAnalysisResult `json:"result"`
|
||
Signals *AISignalBundle `json:"signals"`
|
||
Prompt string `json:"prompt"`
|
||
Run models.AIAnalysisRun `json:"run"`
|
||
Warnings []string `json:"warnings"`
|
||
}
|
||
|
||
const aiSystemPrompt = `你是资深 SRE，负责分析 Loki 日志平台的运行状况。
|
||
|
||
严格规则：
|
||
1. 只能基于我给你的"信号数据"下结论。**不许编造**日志内容、错误信息、时间点或服务名。
|
||
2. 每一条结论都必须在 evidence 里引用信号数据中的具体数字或条目。
|
||
3. 如果信号数据不足以判断根因，把它写进 unknowns，不要猜。
|
||
4. 你看不到日志原文（只有统计与排行），因此不要描述"日志里写了什么"。
|
||
5. 输出必须是**单个 JSON 对象**，不要 markdown 代码块，不要任何解释性前后缀。
|
||
|
||
JSON 结构：
|
||
{
|
||
"summary": "两三句话的整体判断",
|
||
"findings": [{"title": "简短结论", "evidence": "引用具体数据", "severity": "high|medium|low"}],
|
||
"actions": ["可执行的下一步动作，按优先级排序"],
|
||
"unknowns": ["无法从当前数据判断的点"]
|
||
}`
|
||
|
||
// RunAIAnalysis 执行一次分析：采集信号 → 脱敏 → 调模型 → 解析 → 落审计。
|
||
func RunAIAnalysis(ctx context.Context, cfg models.AIConfig, src *models.LogSource, userID uint, username string, admin bool, from, to time.Time, question, app string) (*AIAnalysisOutcome, error) {
|
||
if !cfg.Enabled {
|
||
return nil, fmt.Errorf("AI 分析未启用：请先在「智能运维 → 模型配置」中开启并保存")
|
||
}
|
||
|
||
// ── 每日配额闸门（成本控制第一层）──
|
||
if cfg.DailyTokenCap > 0 {
|
||
used := AITokenUsedToday(userID)
|
||
if used >= cfg.DailyTokenCap {
|
||
run := models.AIAnalysisRun{
|
||
UserID: userID, Username: username, SourceID: src.ID, SourceName: src.Name,
|
||
Status: "quota-exceeded", Error: fmt.Sprintf("今日 AI token 用量已达上限（%d/%d）", used, cfg.DailyTokenCap),
|
||
CreatedAt: time.Now(),
|
||
}
|
||
persistAIRun(&run)
|
||
return nil, fmt.Errorf("今日 AI 用量已达上限（%d/%d tokens），可在模型配置中调整上限或明天再试", used, cfg.DailyTokenCap)
|
||
}
|
||
}
|
||
|
||
started := time.Now()
|
||
signals, err := CollectAISignals(ctx, src, userID, admin, from, to, app)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// 扫描预算闸门：P0 全是索引/快照，实际扫描为 0，这里留出扩展位
|
||
if cfg.MaxScanBytes > 0 && signals.TotalBytes > cfg.MaxScanBytes*8 {
|
||
signals.Warnings = append(signals.Warnings,
|
||
fmt.Sprintf("该范围日志体积 %.1f GB 偏大，已只使用索引统计（未读取日志内容）",
|
||
float64(signals.TotalBytes)/1024/1024/1024))
|
||
}
|
||
|
||
// ── 脱敏（第二层：隐私）──
|
||
rawPrompt := SignalsAsText(signals)
|
||
redacted, invalidRules := RedactText(rawPrompt, cfg.RedactExtra)
|
||
if question = strings.TrimSpace(question); question != "" {
|
||
redacted += "\n\n运维补充的问题：" + RedactTextOnly(question, cfg.RedactExtra)
|
||
}
|
||
warnings := append([]string{}, signals.Warnings...)
|
||
for _, r := range invalidRules {
|
||
warnings = append(warnings, "额外脱敏规则无效已跳过："+r)
|
||
}
|
||
|
||
// ── 调模型 ──
|
||
reply, promptTokens, completionTokens, callErr := aiChat(ctx, cfg, []aiMessage{
|
||
{Role: "system", Content: aiSystemPrompt},
|
||
{Role: "user", Content: redacted},
|
||
})
|
||
|
||
run := models.AIAnalysisRun{
|
||
UserID: userID, Username: username, SourceID: src.ID, SourceName: src.Name,
|
||
RangeMinutes: signals.RangeMinutes, Model: cfg.Model, Question: question,
|
||
PromptPayload: truncateStr(redacted, 20000),
|
||
PromptTokens: promptTokens, CompletionTokens: completionTokens,
|
||
ScanBytes: 0, // P0 不扫内容
|
||
DurationMs: time.Since(started).Milliseconds(),
|
||
CreatedAt: time.Now(),
|
||
}
|
||
if callErr != nil {
|
||
run.Status = "failed"
|
||
run.Error = callErr.Error()
|
||
run.DurationMs = time.Since(started).Milliseconds()
|
||
persistAIRun(&run)
|
||
return nil, callErr
|
||
}
|
||
|
||
result := parseAIResult(reply)
|
||
conclusion, _ := json.Marshal(result)
|
||
run.Status = "ok"
|
||
run.Conclusion = string(conclusion)
|
||
run.DurationMs = time.Since(started).Milliseconds()
|
||
persistAIRun(&run)
|
||
|
||
return &AIAnalysisOutcome{
|
||
Result: result, Signals: signals, Prompt: redacted, Run: run, Warnings: warnings,
|
||
}, nil
|
||
}
|
||
|
||
// RedactTextOnly 只脱敏不做规则校验提示（用于短文本如用户提问）。
|
||
func RedactTextOnly(text string, extra []string) string {
|
||
out, _ := RedactText(text, extra)
|
||
return out
|
||
}
|
||
|
||
// parseAIResult 宽松解析模型输出。
|
||
//
|
||
// 模型常见的"不听话"有三种：包了 ```json 代码块、前后加了解释、
|
||
// 直接返回纯文本。前两种剥壳后重试，第三种降级为 summary=原文，
|
||
// 保证用户至少能看到内容，而不是一个"解析失败"。
|
||
func parseAIResult(reply string) AIAnalysisResult {
|
||
text := strings.TrimSpace(reply)
|
||
text = strings.TrimPrefix(text, "```json")
|
||
text = strings.TrimPrefix(text, "```")
|
||
text = strings.TrimSuffix(text, "```")
|
||
text = strings.TrimSpace(text)
|
||
if idx := strings.Index(text, "{"); idx > 0 {
|
||
text = text[idx:]
|
||
}
|
||
if end := strings.LastIndex(text, "}"); end >= 0 && end < len(text)-1 {
|
||
text = text[:end+1]
|
||
}
|
||
|
||
var parsed AIAnalysisResult
|
||
if err := json.Unmarshal([]byte(text), &parsed); err == nil && (parsed.Summary != "" || len(parsed.Findings) > 0) {
|
||
return parsed
|
||
}
|
||
return AIAnalysisResult{
|
||
Summary: strings.TrimSpace(reply),
|
||
Findings: []AIFinding{},
|
||
Actions: []string{},
|
||
Unknowns: []string{},
|
||
RawText: reply,
|
||
ParseErr: "模型未按 JSON 格式返回，已按纯文本展示",
|
||
}
|
||
}
|
||
|
||
// persistAIRun 落审计（同时用于每日 token 统计）。
|
||
func persistAIRun(run *models.AIAnalysisRun) {
|
||
db := repository.GetDB()
|
||
if db == nil {
|
||
return
|
||
}
|
||
newID := db.NextID("aiAnalysisRuns")
|
||
db.Lock()
|
||
defer db.Unlock()
|
||
run.ID = newID
|
||
db.AIAnalysisRuns = append(db.AIAnalysisRuns, *run)
|
||
// 只保留最近 500 条，避免审计表无限膨胀（明细可通过日志追溯）
|
||
if len(db.AIAnalysisRuns) > 500 {
|
||
db.AIAnalysisRuns = db.AIAnalysisRuns[len(db.AIAnalysisRuns)-500:]
|
||
}
|
||
db.RequestSave()
|
||
}
|
||
|
||
// ListAIRuns 分页查询审计记录（按时间倒序）。
|
||
func ListAIRuns(page, pageSize int) ([]models.AIAnalysisRun, int) {
|
||
if page <= 0 {
|
||
page = 1
|
||
}
|
||
if pageSize <= 0 || pageSize > 100 {
|
||
pageSize = 20
|
||
}
|
||
db := repository.GetDB()
|
||
if db == nil {
|
||
return nil, 0
|
||
}
|
||
db.RLock()
|
||
all := make([]models.AIAnalysisRun, len(db.AIAnalysisRuns))
|
||
copy(all, db.AIAnalysisRuns)
|
||
db.RUnlock()
|
||
|
||
total := len(all)
|
||
// 倒序（新的在前）
|
||
for i, j := 0, len(all)-1; i < j; i, j = i+1, j-1 {
|
||
all[i], all[j] = all[j], all[i]
|
||
}
|
||
start := (page - 1) * pageSize
|
||
if start >= total {
|
||
return []models.AIAnalysisRun{}, total
|
||
}
|
||
end := start + pageSize
|
||
if end > total {
|
||
end = total
|
||
}
|
||
return all[start:end], total
|
||
}
|
||
|
||
// GetAIRun 取单条审计记录（用于查看完整 payload）。
|
||
func GetAIRun(id uint) *models.AIAnalysisRun {
|
||
db := repository.GetDB()
|
||
if db == nil {
|
||
return nil
|
||
}
|
||
db.RLock()
|
||
defer db.RUnlock()
|
||
for i := range db.AIAnalysisRuns {
|
||
if db.AIAnalysisRuns[i].ID == id {
|
||
run := db.AIAnalysisRuns[i]
|
||
return &run
|
||
}
|
||
}
|
||
return nil
|
||
}
|