opsx
5f42ac5486
=== 先测清楚瓶颈在哪 === 后端单点实测（都很快，不是问题所在）： /logs/my-scope 10ms /logs/label/namespace/values 25ms /logs/label-values?scoped astralx 23ms /logs/label/app/values scoped 18ms 结论：慢的是**前端每次都重新走一遍请求链**（切数据源 / 切命名空间 / 重开面板 都会触发），三四次往返串起来 + 每次都有 loading 态，用户看到的就是转圈。 === 后端：label 候选值短缓存 === services/log_query.go：60 秒 TTL，key = sourceId|label|selector|时间窗(按 5 分钟取整)， 上限 500 条并做过期清理。理由：下拉值的时效性要求很低—— 新上线的 app 晚 1 分钟出现在候选里完全可以接受。 LokiLabelValuesBatch 内部逐个走 LokiLabelValues，所以批量接口自动受益。 === 前端：内存缓存 + 飞行中请求合并 === api/logs/index.ts：60 秒 TTL 的 Map，**缓存 Promise 而不是结果**， 所以并发的相同请求共享同一次飞行（去重）；失败立刻剔除缓存，避免缓存错误。 带 AbortSignal 的调用不参与缓存（用户主动取消的请求不应共享给别人）。 额外导出 invalidateLabelCache()，数据源变更时可手动清。 === 预期效果 === 首次打开：与之前相同（~100ms 量级） 再次切换 / 重开面板：直接命中内存，0 请求 服务端重复请求：25~40ms → 微秒级
1056 lines
34 KiB
Go
1056 lines
34 KiB
Go
package services
|
||
|
||
// 日志查询代理 — 后端代表前端访问 Loki HTTP API。
|
||
// cluster-svc 型经 kube-apiserver service proxy(不依赖 ClusterIP 直通与集群 DNS);
|
||
// url 型直连。契约参照 aiops 日志中心(query/labels/values/histogram + cursor 翻页)。
|
||
import (
|
||
"context"
|
||
"crypto/tls"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"net/url"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"k8s.io/client-go/rest"
|
||
"lokistack/models"
|
||
"lokistack/repository"
|
||
)
|
||
|
||
// maxLogQueryLimit 单页硬上限，与前端下拉选项保持**完全一致**，
|
||
// 避免"前端能选 5000、后端静默截成 1000"这种假选项。
|
||
// 取值 5000 = Loki limits_config.max_entries_limit_per_query 的默认上限；
|
||
// 前端结果表已加虚拟滚动（>100 条只渲染可视区），5000 行不会卡。
|
||
const maxLogQueryLimit = 5000
|
||
const logQueryTimeout = 120 * time.Second
|
||
|
||
var lokiHTTPClients = [2]*http.Client{
|
||
{Timeout: logQueryTimeout, Transport: newLokiTransport(false)},
|
||
{Timeout: logQueryTimeout, Transport: newLokiTransport(true)},
|
||
}
|
||
|
||
func newLokiTransport(skipTLSVerify bool) *http.Transport {
|
||
transport := http.DefaultTransport.(*http.Transport).Clone()
|
||
transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: skipTLSVerify} //nolint:gosec // 数据源显式配置
|
||
return transport
|
||
}
|
||
|
||
// ==================== Loki HTTP 访问 ====================
|
||
|
||
// lokiProxyGET 对数据源执行 GET；cluster-svc 使用 client-go 的认证传输访问 API Server Service Proxy。
|
||
func lokiProxyGET(ctx context.Context, src *models.LogSource, path string, query url.Values) ([]byte, int, error) {
|
||
if src.TargetType == "cluster-svc" {
|
||
db := repository.GetDB()
|
||
db.RLock()
|
||
var cluster *models.Cluster
|
||
for i := range db.Clusters {
|
||
if src.ClusterID != nil && db.Clusters[i].ID == *src.ClusterID {
|
||
c := db.Clusters[i]
|
||
cluster = &c
|
||
break
|
||
}
|
||
}
|
||
db.RUnlock()
|
||
if cluster == nil {
|
||
return nil, 500, fmt.Errorf("数据源关联的集群不存在")
|
||
}
|
||
if !kubernetesNamespacePattern.MatchString(src.Namespace) || !kubernetesNamespacePattern.MatchString(src.Service) || src.Port < 1 || src.Port > 65535 {
|
||
return nil, 500, fmt.Errorf("Loki 数据源的命名空间、Service 或端口无效")
|
||
}
|
||
config, err := clusterRESTConfig(cluster, logQueryTimeout)
|
||
if err != nil {
|
||
return nil, 500, err
|
||
}
|
||
client, err := rest.HTTPClientFor(config)
|
||
if err != nil {
|
||
return nil, 500, fmt.Errorf("创建 Kubernetes API 客户端失败: %w", err)
|
||
}
|
||
full := strings.TrimRight(config.Host, "/") + fmt.Sprintf("/api/v1/namespaces/%s/services/%s:%d/proxy%s",
|
||
src.Namespace, src.Service, src.Port, path)
|
||
if qs := query.Encode(); qs != "" {
|
||
full += "?" + qs
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, full, nil)
|
||
if err != nil {
|
||
return nil, 500, err
|
||
}
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
return nil, 502, fmt.Errorf("Kubernetes API 代理查询失败: %w", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
body, err := io.ReadAll(io.LimitReader(resp.Body, 32<<20))
|
||
if err != nil {
|
||
return nil, 502, err
|
||
}
|
||
if resp.StatusCode >= 400 {
|
||
return body, resp.StatusCode, fmt.Errorf("Kubernetes API 代理返回 HTTP %d: %s", resp.StatusCode, truncateStr(string(body), 300))
|
||
}
|
||
return body, resp.StatusCode, nil
|
||
}
|
||
|
||
// url 直连
|
||
if src.URL == "" {
|
||
return nil, 500, fmt.Errorf("数据源未配置访问地址")
|
||
}
|
||
full := strings.TrimSuffix(src.URL, "/") + path
|
||
if qs := query.Encode(); qs != "" {
|
||
full += "?" + qs
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, full, nil)
|
||
if err != nil {
|
||
return nil, 500, err
|
||
}
|
||
applyLokiAuth(req, src)
|
||
client := lokiHTTPClients[0]
|
||
if src.SkipTLSVerify {
|
||
client = lokiHTTPClients[1]
|
||
}
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
return nil, 502, err
|
||
}
|
||
defer resp.Body.Close()
|
||
body, err := io.ReadAll(io.LimitReader(resp.Body, 32<<20))
|
||
if err != nil {
|
||
return nil, 502, err
|
||
}
|
||
if resp.StatusCode >= 400 {
|
||
return body, resp.StatusCode, fmt.Errorf("Loki 返回 HTTP %d: %s", resp.StatusCode, truncateStr(string(body), 300))
|
||
}
|
||
return body, resp.StatusCode, nil
|
||
}
|
||
|
||
func applyLokiAuth(req *http.Request, src *models.LogSource) {
|
||
switch src.AuthType {
|
||
case "basic":
|
||
if pass, err := DecryptCred(src.BasicPass); err == nil {
|
||
req.Header.Set("Authorization", "Basic "+
|
||
base64.StdEncoding.EncodeToString([]byte(src.BasicUser+":"+pass)))
|
||
}
|
||
case "bearer":
|
||
if token, err := DecryptCred(src.BearerToken); err == nil {
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
}
|
||
}
|
||
for name, encrypted := range src.Headers {
|
||
if validSourceHeader(name) {
|
||
if value, err := DecryptCred(encrypted); err == nil {
|
||
req.Header.Set(name, value)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func truncateStr(s string, n int) string {
|
||
if len(s) <= n {
|
||
return s
|
||
}
|
||
return s[:n] + "..."
|
||
}
|
||
|
||
// ==================== 查询实现(契约对齐 aiops) ====================
|
||
|
||
type LogEntryRow struct {
|
||
Timestamp string `json:"timestamp"`
|
||
TimestampNs string `json:"timestampNs"`
|
||
Level string `json:"level"`
|
||
Message string `json:"message"`
|
||
Labels map[string]string `json:"labels"`
|
||
}
|
||
|
||
type LogQueryResult struct {
|
||
Entries []LogEntryRow `json:"entries"`
|
||
HasMore bool `json:"hasMore"`
|
||
NextCursor string `json:"nextCursor"`
|
||
}
|
||
|
||
// ContextLogSelector uses stable stream identity labels. Loki can add per-entry
|
||
// labels such as detected_level to query results; matching those would hide
|
||
// adjacent lines whose detected level differs from the selected line.
|
||
func ContextLogSelector(labels map[string]string) (string, error) {
|
||
for key := range labels {
|
||
if !ValidLogLabelName(key) {
|
||
return "", fmt.Errorf("标签名无效")
|
||
}
|
||
}
|
||
keys := []string{"cluster", "namespace", "app", "pod", "container", "filename"}
|
||
if labels["pod"] == "" {
|
||
keys = append(keys, "job", "instance", "host")
|
||
}
|
||
parts := make([]string, 0, len(keys))
|
||
for _, key := range keys {
|
||
if value := labels[key]; value != "" {
|
||
encoded, _ := json.Marshal(value)
|
||
parts = append(parts, key+"="+string(encoded))
|
||
}
|
||
}
|
||
if len(parts) == 0 {
|
||
for key, value := range labels {
|
||
if value == "" || key == "detected_level" || key == "level" || strings.HasPrefix(key, "__") {
|
||
continue
|
||
}
|
||
encoded, _ := json.Marshal(value)
|
||
parts = append(parts, key+"="+string(encoded))
|
||
}
|
||
sort.Strings(parts)
|
||
}
|
||
if len(parts) == 0 {
|
||
return "", fmt.Errorf("缺少可用于定位日志流的标签")
|
||
}
|
||
return "{" + strings.Join(parts, ",") + "}", nil
|
||
}
|
||
|
||
// QueryLokiLogs 日志查询:flatten streams → 统一行;cursor(最后一条纳秒时间戳)向后翻页
|
||
func QueryLokiLogs(ctx context.Context, src *models.LogSource, query string, start, end int64, limit int, direction, cursor string) (*LogQueryResult, error) {
|
||
if limit <= 0 {
|
||
limit = 200
|
||
}
|
||
if limit > maxLogQueryLimit {
|
||
limit = maxLogQueryLimit
|
||
}
|
||
if direction == "" {
|
||
direction = "backward"
|
||
}
|
||
if end <= 0 {
|
||
end = time.Now().UnixNano()
|
||
}
|
||
if start <= 0 {
|
||
start = end - int64(time.Hour) // 默认最近 1 小时
|
||
}
|
||
if cursor != "" {
|
||
if c, err := strconv.ParseInt(cursor, 10, 64); err == nil {
|
||
if direction == "forward" {
|
||
start = c + 1
|
||
} else {
|
||
end = c - 1
|
||
}
|
||
}
|
||
}
|
||
|
||
q := url.Values{}
|
||
q.Set("query", query)
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
q.Set("limit", strconv.Itoa(limit))
|
||
q.Set("direction", direction)
|
||
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/query_range", q)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var resp struct {
|
||
Status string `json:"status"`
|
||
Data struct {
|
||
ResultType string `json:"resultType"`
|
||
Result []struct {
|
||
Stream map[string]string `json:"stream"`
|
||
Values [][2]string `json:"values"`
|
||
} `json:"result"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return nil, fmt.Errorf("解析 Loki 响应失败: %w", err)
|
||
}
|
||
|
||
entries := make([]LogEntryRow, 0, limit)
|
||
var lastTs string
|
||
for _, stream := range resp.Data.Result {
|
||
for _, v := range stream.Values {
|
||
entries = append(entries, LogEntryRow{
|
||
Timestamp: formatLokiTimestamp(v[0]),
|
||
TimestampNs: v[0],
|
||
Level: detectLevel(stream.Stream, v[1]),
|
||
Message: v[1],
|
||
Labels: stream.Stream,
|
||
})
|
||
}
|
||
}
|
||
sort.SliceStable(entries, func(i, j int) bool {
|
||
a, _ := strconv.ParseInt(entries[i].TimestampNs, 10, 64)
|
||
b, _ := strconv.ParseInt(entries[j].TimestampNs, 10, 64)
|
||
if direction == "forward" {
|
||
return a < b
|
||
}
|
||
return a > b
|
||
})
|
||
if len(entries) > limit {
|
||
entries = entries[:limit]
|
||
}
|
||
if len(entries) > 0 {
|
||
lastTs = entries[len(entries)-1].TimestampNs
|
||
}
|
||
|
||
nextCursor := ""
|
||
if len(entries) >= limit && lastTs != "" {
|
||
if ts, err := strconv.ParseInt(strings.Split(lastTs, ".")[0], 10, 64); err == nil {
|
||
nextCursor = strconv.FormatInt(ts, 10)
|
||
}
|
||
}
|
||
return &LogQueryResult{Entries: entries, HasMore: nextCursor != "", NextCursor: nextCursor}, nil
|
||
}
|
||
|
||
// formatLokiTimestamp Loki 时间串(纳秒字符串/科学计数法)→ RFC3339
|
||
func formatLokiTimestamp(ts string) string {
|
||
if i := strings.IndexAny(ts, ".eE"); i > 0 {
|
||
ts = ts[:i]
|
||
}
|
||
n, err := strconv.ParseInt(ts, 10, 64)
|
||
if err != nil {
|
||
return ts
|
||
}
|
||
// 自适应纳秒/微秒/毫秒/秒
|
||
switch {
|
||
case n > 1e18:
|
||
return time.Unix(0, n).Format(time.RFC3339)
|
||
case n > 1e15:
|
||
return time.Unix(0, n*1e3).Format(time.RFC3339)
|
||
case n > 1e12:
|
||
return time.Unix(0, n*1e6).Format(time.RFC3339)
|
||
case n > 1e9:
|
||
return time.Unix(0, n*1e9).Format(time.RFC3339)
|
||
default:
|
||
return time.Unix(n, 0).Format(time.RFC3339)
|
||
}
|
||
}
|
||
|
||
// detectLevel 优先取 level 标签;否则按行内容推断
|
||
func detectLevel(labels map[string]string, line string) string {
|
||
if lv, ok := labels["level"]; ok && lv != "" {
|
||
return strings.ToLower(lv)
|
||
}
|
||
l := strings.ToLower(line)
|
||
for _, p := range []string{"error", "err", "fatal", "panic"} {
|
||
if strings.Contains(l, p) {
|
||
return "error"
|
||
}
|
||
}
|
||
if strings.Contains(l, "warn") {
|
||
return "warn"
|
||
}
|
||
return "info"
|
||
}
|
||
|
||
// LokiLabels 标签名列表
|
||
func LokiLabels(ctx context.Context, src *models.LogSource, start, end int64) ([]string, error) {
|
||
q := url.Values{}
|
||
if end > 0 {
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
}
|
||
if start > 0 {
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
}
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/labels", q)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var resp struct {
|
||
Data []string `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return nil, err
|
||
}
|
||
// Loki 在某 label 无值时会返回 {"data":null}，Go nil slice 序列化成 JSON null，
|
||
// 前端 setState(null) 后 .map() 直接抛 "Cannot read properties of null"。
|
||
// 统一归一化为空切片，前端拿到的永远是数组。
|
||
if resp.Data == nil {
|
||
resp.Data = []string{}
|
||
}
|
||
return resp.Data, nil
|
||
}
|
||
|
||
// LokiLabelValues 标签值列表(query 只取 {...} selector 部分)
|
||
// labelValuesDefaultWindow 是 label 值下拉框的默认时间窗。
|
||
// 只影响"没显式给时间范围"的调用（下拉框），不影响带范围的查询。
|
||
const labelValuesDefaultWindow = time.Hour
|
||
|
||
// ── label 候选值的短缓存 ────────────────────────────────────────────
|
||
// 快捷 LogQL 的下拉框在「切数据源 / 切命名空间 / 开关面板」时都会重新请求，
|
||
// 而值本身的时效性要求很低（新上线的 app 晚 1 分钟出现完全可接受）。
|
||
// 缓存后重复请求从 25~40ms 降到微秒级，体感是"打开就有"。
|
||
const (
|
||
labelCacheTTL = 60 * time.Second
|
||
labelCacheMax = 500
|
||
)
|
||
|
||
type labelCacheEntry struct {
|
||
values []string
|
||
expiresAt time.Time
|
||
}
|
||
|
||
var (
|
||
labelCacheMu sync.Mutex
|
||
labelCache = map[string]labelCacheEntry{}
|
||
)
|
||
|
||
func labelCacheKey(src *models.LogSource, name, selector string, start, end int64) string {
|
||
if src == nil {
|
||
return ""
|
||
}
|
||
// 时间窗按 5 分钟取整：避免"每次请求差几秒"导致缓存全不命中
|
||
bucket := int64(5 * time.Minute)
|
||
return fmt.Sprintf("%d|%s|%s|%d|%d", src.ID, name, selector, start/bucket, end/bucket)
|
||
}
|
||
|
||
func labelCacheGet(key string) ([]string, bool) {
|
||
if key == "" {
|
||
return nil, false
|
||
}
|
||
labelCacheMu.Lock()
|
||
defer labelCacheMu.Unlock()
|
||
if entry, ok := labelCache[key]; ok && time.Now().Before(entry.expiresAt) {
|
||
return entry.values, true
|
||
}
|
||
return nil, false
|
||
}
|
||
|
||
func labelCachePut(key string, values []string) {
|
||
if key == "" {
|
||
return
|
||
}
|
||
labelCacheMu.Lock()
|
||
defer labelCacheMu.Unlock()
|
||
if len(labelCache) >= labelCacheMax {
|
||
now := time.Now()
|
||
for k, v := range labelCache {
|
||
if now.After(v.expiresAt) {
|
||
delete(labelCache, k)
|
||
}
|
||
}
|
||
if len(labelCache) >= labelCacheMax {
|
||
labelCache = map[string]labelCacheEntry{}
|
||
}
|
||
}
|
||
labelCache[key] = labelCacheEntry{values: values, expiresAt: time.Now().Add(labelCacheTTL)}
|
||
}
|
||
|
||
func LokiLabelValues(ctx context.Context, src *models.LogSource, name, selector string, start, end int64) ([]string, error) {
|
||
// 缓存按调用方传入的原始 start/end 计算 key（下方默认窗口不影响 key 的稳定性）
|
||
cacheKey := labelCacheKey(src, name, selector, start, end)
|
||
if cached, ok := labelCacheGet(cacheKey); ok {
|
||
return cached, nil
|
||
}
|
||
q := url.Values{}
|
||
if s := extractStreamSelector(selector); s != "" && s != "{}" {
|
||
q.Set("query", s)
|
||
}
|
||
// 不传时间范围时 Loki 默认扫**最近 6 小时**的索引 —— 在 370GB/天的集群上
|
||
// 实测要 10~21 秒（下拉框"一直加载中"的直接原因）。
|
||
// 下拉框只需要"现在有哪些 app/pod 在打日志"，给 1 小时足够，索引量降到 1/6。
|
||
// 调用方显式传了时间范围则完全尊从，不受影响。
|
||
if start <= 0 && end <= 0 {
|
||
now := time.Now()
|
||
start = now.Add(-labelValuesDefaultWindow).UnixNano()
|
||
end = now.UnixNano()
|
||
}
|
||
if end > 0 {
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
}
|
||
if start > 0 {
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
}
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/label/"+url.PathEscape(name)+"/values", q)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var resp struct {
|
||
Data []string `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return nil, err
|
||
}
|
||
// 同上：label 无值时 Loki 返回 {"data":null}，归一化为空切片，
|
||
// 避免前端 .map() 遇到 null 崩溃（快捷 LogQL 报错的直接原因）。
|
||
if resp.Data == nil {
|
||
resp.Data = []string{}
|
||
}
|
||
labelCachePut(cacheKey, resp.Data)
|
||
return resp.Data, nil
|
||
}
|
||
|
||
// LokiLabelValuesBatch 一次取多个 label 的候选值。
|
||
//
|
||
// 前端快捷查询要对 app / service_name / container 做逐级回退（哪个有值用哪个），
|
||
// 原来最多并发 3 个请求、每个请求各走一次 Loki 索引查询 + 一次 kube-apiserver
|
||
// service proxy；合并成本接口后前端只发一次，后端内部并发（上限 4）。
|
||
//
|
||
// 语义：单个 label 失败不影响其它 label（返回空数组），**全部失败**才报错 ——
|
||
// 否则某个 label 在特定 selector 下索引异常，整个下拉框都会变成"加载失败"。
|
||
func LokiLabelValuesBatch(ctx context.Context, src *models.LogSource, names []string, selector string, start, end int64) (map[string][]string, error) {
|
||
out := make(map[string][]string, len(names))
|
||
var (
|
||
mu sync.Mutex
|
||
wg sync.WaitGroup
|
||
firstErr error
|
||
okCount int
|
||
seen = make(map[string]bool, len(names))
|
||
)
|
||
sem := make(chan struct{}, 4)
|
||
for _, raw := range names {
|
||
name := strings.TrimSpace(raw)
|
||
if name == "" || seen[name] {
|
||
continue
|
||
}
|
||
seen[name] = true
|
||
out[name] = []string{}
|
||
wg.Add(1)
|
||
go func(labelName string) {
|
||
defer wg.Done()
|
||
sem <- struct{}{}
|
||
defer func() { <-sem }()
|
||
values, err := LokiLabelValues(ctx, src, labelName, selector, start, end)
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
if err != nil {
|
||
if firstErr == nil {
|
||
firstErr = err
|
||
}
|
||
return
|
||
}
|
||
okCount++
|
||
out[labelName] = values
|
||
}(name)
|
||
}
|
||
wg.Wait()
|
||
if okCount == 0 && firstErr != nil {
|
||
return out, firstErr
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// LokiVolumeBytesFor 用 index/volume 读某个 selector 的日志体积（字节）。
|
||
//
|
||
// 这是 Loki 的**索引接口**：只读 TSDB 里记录的 chunk 字节数，不解码日志内容。
|
||
// 实测单个 app、一小时窗口约 80ms（对比 bytes_over_time 同窗口 13s/15 分钟、
|
||
// 约 50s/小时 —— 因为它要把整小时的日志解压出来数行字节）。
|
||
//
|
||
// 代价：index/volume 返回的 metric 只包含 selector 里出现过的 label，
|
||
// 所以"按 app 取体积"没法一条查询拿全，必须一个 app 调一次（见 app_volume.go）。
|
||
// aggregateBy=series 让返回值带上 selector 里的全部 label，便于调用方核对。
|
||
func LokiVolumeBytesFor(ctx context.Context, src *models.LogSource, selector string, start, end int64) (int64, error) {
|
||
if end <= 0 {
|
||
end = time.Now().UnixNano()
|
||
}
|
||
if start <= 0 {
|
||
start = end - int64(time.Hour)
|
||
}
|
||
ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||
defer cancel()
|
||
|
||
q := url.Values{}
|
||
q.Set("query", selector)
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
q.Set("aggregateBy", "series")
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/index/volume", q)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
var resp struct {
|
||
Data struct {
|
||
Result []struct {
|
||
Metric map[string]string `json:"metric"`
|
||
Value [2]interface{} `json:"value"`
|
||
} `json:"result"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return 0, fmt.Errorf("解析 volume 响应失败: %w", err)
|
||
}
|
||
var total int64
|
||
for _, item := range resp.Data.Result {
|
||
switch v := item.Value[1].(type) {
|
||
case float64:
|
||
total += int64(v)
|
||
case string:
|
||
if n, convErr := strconv.ParseFloat(v, 64); convErr == nil {
|
||
total += int64(n)
|
||
}
|
||
}
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
// extractStreamSelector 从 LogQL 中提取 {...} 部分(labels API 仅接受 stream selector)
|
||
func extractStreamSelector(query string) string {
|
||
i := strings.Index(query, "{")
|
||
if i < 0 {
|
||
return ""
|
||
}
|
||
j := strings.Index(query[i:], "}")
|
||
if j < 0 {
|
||
return ""
|
||
}
|
||
return query[i : i+j+1]
|
||
}
|
||
|
||
// LokiHistogram 日志量直方图:服务端包 sum(count_over_time(q[step]))
|
||
// 重查询护栏。设计依据：
|
||
// - 单条 27 天的 count_over_time 曾把 Loki 打到 OOMKilled(2Gi)，
|
||
// 进而 Pod 不 Ready → Service 无端点 → 全站查询 502。
|
||
// - 所以"算不完的查询"必须在入口就拒绝，而不是硬跑 120 秒把实例拖垮。
|
||
const (
|
||
// 日志量统计最多统计 7 天；更长跨度请分段查（前端会提示缩小范围）
|
||
maxHistogramRange = 7 * 24 * time.Hour
|
||
// 重查询（histogram/metrics）单独的更短超时，避免长时间占用 CPU
|
||
// 90s：实测 astralx（约 370GB/天）15 分钟窗口单条 count_over_time
|
||
// exec 约 29s、外加排队时间，30s 必超时 → 分析页 502。
|
||
// 索引类查询（index/volume_range、index/stats）不受此限制，仍是毫秒级。
|
||
heavyQueryTimeout = 90 * time.Second
|
||
// 全局并发重查询上限。取 4 而不是 2：
|
||
// 日志分析页一次会并发 5 条 count_over_time，限制 2 会让大部分请求拿不到令牌。
|
||
maxConcurrentHeavyQueries = 4
|
||
// 拿不到令牌时最多排队 10 秒再拒绝。
|
||
// 原来"立即 429"会把分析页 5 条并发里的 3 条直接打回，
|
||
// 用户看到的就是"当前有较多日志量统计在进行"。
|
||
heavyQueryWait = 10 * time.Second
|
||
)
|
||
|
||
// ErrTooManyQueries 并发重查询超限（handler 映射为 429）
|
||
var ErrTooManyQueries = errors.New("当前有较多日志量统计在进行，请稍后重试")
|
||
|
||
// heavyQuerySem 全局重查询信号量
|
||
var heavyQuerySem = make(chan struct{}, maxConcurrentHeavyQueries)
|
||
|
||
// acquireHeavyQuery 先尝试立即获取；拿不到则最多排队 heavyQueryWait。
|
||
// 突发（如分析页并发 5 条）会自然排队消化，只有持续超载才返回 429。
|
||
func acquireHeavyQuery() bool {
|
||
select {
|
||
case heavyQuerySem <- struct{}{}:
|
||
return true
|
||
default:
|
||
}
|
||
timer := time.NewTimer(heavyQueryWait)
|
||
defer timer.Stop()
|
||
select {
|
||
case heavyQuerySem <- struct{}{}:
|
||
return true
|
||
case <-timer.C:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func releaseHeavyQuery() { <-heavyQuerySem }
|
||
|
||
func LokiHistogram(ctx context.Context, src *models.LogSource, query string, start, end int64) ([]map[string]interface{}, error) {
|
||
if end <= 0 {
|
||
end = time.Now().UnixNano()
|
||
}
|
||
if start <= 0 {
|
||
start = end - int64(time.Hour)
|
||
}
|
||
span := end - start
|
||
if span > int64(maxHistogramRange) {
|
||
return nil, fmt.Errorf("统计范围过大（%.1f 天），最多支持 %.0f 天；请缩小时间范围后重试",
|
||
float64(span)/float64(24*time.Hour), maxHistogramRange.Hours()/24)
|
||
}
|
||
step := pickHistogramStep(span)
|
||
|
||
// ── 首选：Loki 官方的 index/volume_range ──────────────────────────────
|
||
// 它只读 TSDB 索引里的 chunk 元数据（字节数），**不解码日志内容**。
|
||
// 实测同一条件 24 小时：count_over_time 53s 超时失败，volume_range 111ms。
|
||
// 返回单位是「字节」，所以 bucket 里带 unit=bytes，前端据此格式化为 KB/MB/GB。
|
||
//
|
||
// ⚠️ 这条路径**不占用重型查询令牌**：
|
||
// 索引查询本身是毫秒级的，如果和 count_over_time 之类要读 chunk 的查询抢
|
||
// 同一个信号量，就会出现"图表明明 0 秒能出，却因为别人在跑分析查询而排队
|
||
// 十秒、最后 429/超时"的怪现象（分析页实际踩过）。
|
||
q := url.Values{}
|
||
q.Set("query", query)
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
q.Set("step", step)
|
||
q.Set("limit", "50")
|
||
idxCtx, idxCancel := context.WithTimeout(ctx, 15*time.Second)
|
||
if body, status, err := lokiProxyGET(idxCtx, src, "/loki/api/v1/index/volume_range", q); err == nil && status < 400 {
|
||
if buckets, ok := parseVolumeBuckets(body); ok {
|
||
idxCancel()
|
||
return buckets, nil
|
||
}
|
||
}
|
||
idxCancel()
|
||
|
||
// ── 回退：老版本 Loki 没有 volume_range 时才走 count_over_time ──
|
||
// 这条要真读 chunk，才需要抢重型令牌 + 长超时。
|
||
if !acquireHeavyQuery() {
|
||
return nil, ErrTooManyQueries
|
||
}
|
||
defer releaseHeavyQuery()
|
||
heavyCtx, cancel := context.WithTimeout(ctx, heavyQueryTimeout)
|
||
defer cancel()
|
||
return lokiHistogramByCount(heavyCtx, src, query, step, start, end)
|
||
}
|
||
|
||
// parseVolumeBuckets 解析 index/volume_range 的 matrix 响应。
|
||
// 成功但无数据也返回 ok=true（空图），只有结构不对才 ok=false 触发回退。
|
||
func parseVolumeBuckets(body []byte) ([]map[string]interface{}, bool) {
|
||
var resp struct {
|
||
Data struct {
|
||
Result []struct {
|
||
Values [][2]interface{} `json:"values"`
|
||
} `json:"result"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return nil, false
|
||
}
|
||
buckets := []map[string]interface{}{}
|
||
if len(resp.Data.Result) > 0 {
|
||
for _, v := range resp.Data.Result[0].Values {
|
||
tsf, _ := strconv.ParseFloat(fmt.Sprintf("%v", v[0]), 64)
|
||
bytes, _ := strconv.ParseFloat(fmt.Sprintf("%v", v[1]), 64)
|
||
buckets = append(buckets, map[string]interface{}{
|
||
"timestamp": time.Unix(int64(tsf), 0).Format(time.RFC3339),
|
||
"count": int64(bytes),
|
||
"unit": "bytes",
|
||
})
|
||
}
|
||
}
|
||
return buckets, true
|
||
}
|
||
|
||
// lokiHistogramByCount 回退实现：count_over_time 逐点统计「条数」。
|
||
// 代价高（要真读 chunk），只在 volume_range 不可用时使用。
|
||
func lokiHistogramByCount(ctx context.Context, src *models.LogSource, query, step string, start, end int64) ([]map[string]interface{}, error) {
|
||
wrapped := fmt.Sprintf("sum(count_over_time(%s[%s]))", query, step)
|
||
q := url.Values{}
|
||
q.Set("query", wrapped)
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
q.Set("step", step)
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/query_range", q)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var resp struct {
|
||
Data struct {
|
||
Result []struct {
|
||
Values [][2]interface{} `json:"values"`
|
||
} `json:"result"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return nil, err
|
||
}
|
||
buckets := []map[string]interface{}{}
|
||
if len(resp.Data.Result) > 0 {
|
||
for _, v := range resp.Data.Result[0].Values {
|
||
tsf, _ := strconv.ParseFloat(fmt.Sprintf("%v", v[0]), 64)
|
||
cnt, _ := strconv.ParseFloat(fmt.Sprintf("%v", v[1]), 64)
|
||
buckets = append(buckets, map[string]interface{}{
|
||
"timestamp": time.Unix(int64(tsf), 0).Format(time.RFC3339),
|
||
"count": int64(cnt),
|
||
"unit": "entries",
|
||
})
|
||
}
|
||
}
|
||
return buckets, nil
|
||
}
|
||
|
||
// LokiStats 用官方 /loki/api/v1/index/stats 取「条数/字节/块数」。
|
||
//
|
||
// 为什么不数日志条数：index/stats 只读 TSDB 索引里的 chunk 元数据，
|
||
// 不解码日志内容。实测同一 app 24 小时：count_over_time 53s 失败，
|
||
// index/stats 1.3s。用它来算「共 N 条 / 共 M 页」成本可以忽略。
|
||
type LokiStats struct {
|
||
Streams int64 `json:"streams"`
|
||
Chunks int64 `json:"chunks"`
|
||
Bytes int64 `json:"bytes"`
|
||
Entries int64 `json:"entries"`
|
||
}
|
||
|
||
func LokiStatsFor(ctx context.Context, src *models.LogSource, query string, start, end int64) (*LokiStats, error) {
|
||
if end <= 0 {
|
||
end = time.Now().UnixNano()
|
||
}
|
||
if start <= 0 {
|
||
start = end - int64(time.Hour)
|
||
}
|
||
ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||
defer cancel()
|
||
|
||
q := url.Values{}
|
||
q.Set("query", query)
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/index/stats", q)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var st LokiStats
|
||
if err := json.Unmarshal(body, &st); err != nil {
|
||
return nil, err
|
||
}
|
||
return &st, nil
|
||
}
|
||
|
||
// MetricSeriesPoint 指标序列上的一个采样点
|
||
type MetricSeriesPoint struct {
|
||
Timestamp string `json:"timestamp"`
|
||
Value float64 `json:"value"`
|
||
}
|
||
|
||
// MetricSeries 一条按标签分组的指标序列(如 sum by (app)(...) 的每个 app 一条)
|
||
type MetricSeries struct {
|
||
Labels map[string]string `json:"labels"`
|
||
Points []MetricSeriesPoint `json:"points"`
|
||
}
|
||
|
||
// LokiMetrics 通用 LogQL 指标查询(日志分析页用):query 为完整指标表达式,
|
||
// 其中所有 [$step] 占位符替换为按时间跨度自适应的步长后走 query_range,
|
||
// 返回矩阵多序列(直方图只取单序列,这里按 by (...) 分组整表返回)
|
||
// ── 分析页指标查询的短缓存 ────────────────────────────────────────────
|
||
//
|
||
// 为什么需要：`sum by (app)(count_over_time(...))` 在这台集群上扫 15 分钟
|
||
// astralx（约 6.6GB）实测 16~28 秒，Loki 的结果缓存也不命中（请求的时间窗
|
||
// 每次都在变）。用户点刷新、切页、换 tab 都要再等半分钟。
|
||
//
|
||
// 做法：按 (数据源, 查询, 时间窗) 缓存 90 秒，时间窗按 **2 分钟取整**再当 key。
|
||
// 代价是命中缓存时数据最多旧 2 分钟 × 90 秒 TTL —— 对"日志量趋势/排行"这种
|
||
// 聚合视图完全可接受，换来的是重复访问从 ~26s 降到毫秒级。
|
||
const (
|
||
metricCacheTTL = 90 * time.Second
|
||
metricCacheMax = 200
|
||
// 时间窗取整粒度：key 用取整后的值，避免"每次点刷新差几秒就穿透"
|
||
metricCacheBucket = 2 * time.Minute
|
||
)
|
||
|
||
type metricCacheEntry struct {
|
||
series []MetricSeries
|
||
expiresAt time.Time
|
||
}
|
||
|
||
var (
|
||
metricCacheMu sync.Mutex
|
||
metricCache = map[string]metricCacheEntry{}
|
||
)
|
||
|
||
// LokiMetrics 带短缓存的分析指标查询（语义见 metricCache 注释）。
|
||
func LokiMetrics(ctx context.Context, src *models.LogSource, query string, start, end int64) ([]MetricSeries, error) {
|
||
key := ""
|
||
if src != nil {
|
||
bucket := int64(metricCacheBucket)
|
||
key = fmt.Sprintf("%d|%s|%d|%d", src.ID, query, start/bucket, end/bucket)
|
||
metricCacheMu.Lock()
|
||
if entry, ok := metricCache[key]; ok && time.Now().Before(entry.expiresAt) {
|
||
metricCacheMu.Unlock()
|
||
return entry.series, nil
|
||
}
|
||
metricCacheMu.Unlock()
|
||
}
|
||
|
||
series, err := lokiMetricsUncached(ctx, src, query, start, end)
|
||
if err != nil || key == "" {
|
||
return series, err
|
||
}
|
||
|
||
metricCacheMu.Lock()
|
||
if len(metricCache) >= metricCacheMax {
|
||
now := time.Now()
|
||
for k, v := range metricCache {
|
||
if now.After(v.expiresAt) {
|
||
delete(metricCache, k)
|
||
}
|
||
}
|
||
// 清完还满：整体重置，避免无界增长把内存吃掉
|
||
if len(metricCache) >= metricCacheMax {
|
||
metricCache = map[string]metricCacheEntry{}
|
||
}
|
||
}
|
||
metricCache[key] = metricCacheEntry{series: series, expiresAt: time.Now().Add(metricCacheTTL)}
|
||
metricCacheMu.Unlock()
|
||
return series, nil
|
||
}
|
||
|
||
// ── 总条数/总体积的短缓存 ─────────────────────────────────────────────
|
||
// /logs/stats 走 index/stats，实测 5 分钟窗口要 1.2 秒（60 分钟更慢）。
|
||
// 分析页每次进页面、每次切时间范围都会打它，而这类聚合值的可容忍陈旧度
|
||
// 与指标查询一致（分钟级），所以用同样的 90s TTL + 2 分钟取整。
|
||
const (
|
||
statsCacheTTL = 90 * time.Second
|
||
statsCacheMax = 200
|
||
)
|
||
|
||
type statsCacheEntry struct {
|
||
stats *LokiStats
|
||
expiresAt time.Time
|
||
}
|
||
|
||
var (
|
||
statsCacheMu sync.Mutex
|
||
statsCache = map[string]statsCacheEntry{}
|
||
)
|
||
|
||
// LokiStatsCached 带短缓存的总量统计。
|
||
func LokiStatsCached(ctx context.Context, src *models.LogSource, query string, start, end int64) (*LokiStats, error) {
|
||
key := ""
|
||
if src != nil {
|
||
bucket := int64(metricCacheBucket)
|
||
key = fmt.Sprintf("%d|%s|%d|%d", src.ID, query, start/bucket, end/bucket)
|
||
statsCacheMu.Lock()
|
||
if entry, ok := statsCache[key]; ok && time.Now().Before(entry.expiresAt) {
|
||
statsCacheMu.Unlock()
|
||
return entry.stats, nil
|
||
}
|
||
statsCacheMu.Unlock()
|
||
}
|
||
|
||
st, err := LokiStatsFor(ctx, src, query, start, end)
|
||
if err != nil || key == "" {
|
||
return st, err
|
||
}
|
||
|
||
statsCacheMu.Lock()
|
||
if len(statsCache) >= statsCacheMax {
|
||
now := time.Now()
|
||
for k, v := range statsCache {
|
||
if now.After(v.expiresAt) {
|
||
delete(statsCache, k)
|
||
}
|
||
}
|
||
if len(statsCache) >= statsCacheMax {
|
||
statsCache = map[string]statsCacheEntry{}
|
||
}
|
||
}
|
||
statsCache[key] = statsCacheEntry{stats: st, expiresAt: time.Now().Add(statsCacheTTL)}
|
||
statsCacheMu.Unlock()
|
||
return st, nil
|
||
}
|
||
|
||
// lokiMetricsUncached 真正打 Loki 的实现（不带缓存）。
|
||
func lokiMetricsUncached(ctx context.Context, src *models.LogSource, query string, start, end int64) ([]MetricSeries, error) {
|
||
if end <= 0 {
|
||
end = time.Now().UnixNano()
|
||
}
|
||
if start <= 0 {
|
||
start = end - int64(time.Hour)
|
||
}
|
||
// 与 histogram 共用护栏：并发限 2、单次 30s 超时，避免 analysis 页
|
||
// 多个卡片同时发起 count_over_time 把单实例 Loki 打满。
|
||
if !acquireHeavyQuery() {
|
||
return nil, ErrTooManyQueries
|
||
}
|
||
defer releaseHeavyQuery()
|
||
ctx, cancel := context.WithTimeout(ctx, heavyQueryTimeout)
|
||
defer cancel()
|
||
|
||
step := pickHistogramStep(end - start)
|
||
wrapped := strings.ReplaceAll(query, "[$step]", "["+step+"]")
|
||
|
||
q := url.Values{}
|
||
q.Set("query", wrapped)
|
||
q.Set("start", strconv.FormatInt(start, 10))
|
||
q.Set("end", strconv.FormatInt(end, 10))
|
||
q.Set("step", step)
|
||
body, _, err := lokiProxyGET(ctx, src, "/loki/api/v1/query_range", q)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var resp struct {
|
||
Data struct {
|
||
Result []struct {
|
||
Metric map[string]string `json:"metric"`
|
||
Values [][2]interface{} `json:"values"`
|
||
} `json:"result"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &resp); err != nil {
|
||
return nil, err
|
||
}
|
||
series := make([]MetricSeries, 0, len(resp.Data.Result))
|
||
for _, item := range resp.Data.Result {
|
||
points := make([]MetricSeriesPoint, 0, len(item.Values))
|
||
for _, v := range item.Values {
|
||
// metric API 的 ts 为秒级浮点数(如 1.79e+09)
|
||
tsf, _ := strconv.ParseFloat(fmt.Sprintf("%v", v[0]), 64)
|
||
cnt, _ := strconv.ParseFloat(fmt.Sprintf("%v", v[1]), 64)
|
||
points = append(points, MetricSeriesPoint{
|
||
Timestamp: time.Unix(int64(tsf), 0).Format(time.RFC3339),
|
||
Value: cnt,
|
||
})
|
||
}
|
||
if len(points) == 0 {
|
||
continue
|
||
}
|
||
series = append(series, MetricSeries{Labels: item.Metric, Points: points})
|
||
}
|
||
return series, nil
|
||
}
|
||
|
||
// pickHistogramStep 按时间跨度自适应步长(aiops 同款)
|
||
func pickHistogramStep(spanNano int64) string {
|
||
span := time.Duration(spanNano)
|
||
switch {
|
||
case span <= 15*time.Minute:
|
||
return "30s"
|
||
case span <= time.Hour:
|
||
return "1m"
|
||
case span <= 6*time.Hour:
|
||
return "5m"
|
||
case span <= 12*time.Hour:
|
||
return "15m"
|
||
case span <= 7*24*time.Hour:
|
||
// 12 小时以上统一 1 小时一桶。
|
||
// 24 小时原来是 15m（96 根柱子）：横轴挤成一片、且每根柱子只有十来分钟的数据，
|
||
// 看不出趋势；1h → 24 根，和"按小时看日志量"的心智一致。
|
||
return "1h"
|
||
default:
|
||
return "6h"
|
||
}
|
||
}
|
||
|
||
// DeploymentReady 部署级就绪探测:按部署名匹配自动注册的数据源,经其访问 Loki /ready
|
||
func DeploymentReady(ctx context.Context, deploymentName string) (bool, string) {
|
||
if deploymentName == "" {
|
||
return false, ""
|
||
}
|
||
db := repository.GetDB()
|
||
db.RLock()
|
||
var src *models.LogSource
|
||
for i := range db.LogSources {
|
||
if db.LogSources[i].Name == deploymentName {
|
||
s := db.LogSources[i]
|
||
src = &s
|
||
break
|
||
}
|
||
}
|
||
db.RUnlock()
|
||
if src == nil {
|
||
return false, "未注册数据源"
|
||
}
|
||
if LokiReady(ctx, src) {
|
||
return true, ""
|
||
}
|
||
return false, "Loki 未就绪"
|
||
}
|
||
|
||
// LokiReady Loki /ready 探活(部署卡片/查询页状态点)
|
||
func LokiReady(ctx context.Context, src *models.LogSource) bool {
|
||
body, code, err := lokiProxyGET(ctx, src, "/ready", nil)
|
||
if err != nil || code != 200 {
|
||
return false
|
||
}
|
||
return strings.Contains(strings.ToLower(string(body)), "ready")
|
||
}
|
||
|
||
// TestLogSource 数据源连通性(/loki/api/v1/labels 探活)
|
||
func TestLogSource(ctx context.Context, src *models.LogSource) map[string]interface{} {
|
||
labels, err := LokiLabels(ctx, src, 0, 0)
|
||
if err != nil {
|
||
return map[string]interface{}{"success": false, "message": err.Error()}
|
||
}
|
||
return map[string]interface{}{
|
||
"success": true,
|
||
"message": fmt.Sprintf("连接成功,发现 %d 个标签", len(labels)),
|
||
"labels": labels,
|
||
}
|
||
}
|