opsx
4dda6e5ba8
=== 一、快捷 LogQL 报 Cannot read properties of null (reading 'map') ===
根因：Loki 对无值的 label 返回 {"data":null}，Go nil slice 序列化成 JSON null，
前端 setState(null) 后再 .map() / [...null] / .filter() 直接抛错。
表现为「不稳定」——切到某个没有该 label 值的范围就崩。
修复（前后端双保险）：
后端 LokiLabels / LokiLabelValues 归一化 nil → []string{}
（LokiHistogram 原本就返回空切片，无需改）
前端 6 处调用点加 Array.isArray 防御：
QuickQueryTemplates ×3、QueryInput ×2、LokiLabelBrowser ×2
=== 二、选 astralx 只显示 2 个 app（不是权限问题）===
实测：
astralx 117 个 Deployment / 497 个 Pod
带 app 标签的 Pod 只有 2 个（cex-promotion-web / cex-web-h5）
其余 115 个 Deployment 的 Pod 模板没打 app，Kuboard 打的是 k8s.kuboard.cn/name
Loki 里 service_name / container 各有 79 个值（Promtail 从容器名生成）
修复：QuickQueryTemplates 的「app」候选改为三级回退 app → service_name → container，
并用 appLabel 状态记住实际生效的标签名，pod 收窄与 generate() 都用它，
保证生成的 LogQL selector 真能查到数据。
=== 三、查询授权：支持「所有人」+ 内置默认策略 ===
1. 后端 policyGroupAccess：userIds 里的 0 作为哨兵值表示「所有人」(anyone / *)，
匹配任意用户（原来只能逐个指定用户 ID）
2. 种子 seedDefaultPolicy()：首次启动且尚无任何策略组时，创建
「默认策略（所有人 · default 命名空间）」（labels: namespace=default，userIds: [0]）。
管理员可在「查询授权」里把命名空间改成实际业务命名空间（如 astralx），
对该组所有用户立即生效。
3. 前端 PolicyGroupsPage：新增「所有人（anyone）」一键按钮（与具体用户互斥），
列表与编辑区显示「所有人」，校验文案同步更新。
验证：
GET /logs/label/nonexistent_xyz/values → {"data":[]}
GET /logs/label/service_name/values?query={namespace="astralx"} → 79 个
GET /logs/policy-groups → 1 条：默认策略（所有人 · default 命名空间）所有人
镜像：lghost/lokistack:latest + v0.1 = sha256:88a4ba626f62
517 lines
15 KiB
Go
517 lines
15 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"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"net/url"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"opsxloki/models"
|
||
"opsxloki/repository"
|
||
"k8s.io/client-go/rest"
|
||
)
|
||
|
||
const maxLogQueryLimit = 1000
|
||
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 部分)
|
||
func LokiLabelValues(ctx context.Context, src *models.LogSource, name, selector string, start, end int64) ([]string, error) {
|
||
q := url.Values{}
|
||
if s := extractStreamSelector(selector); s != "" && s != "{}" {
|
||
q.Set("query", s)
|
||
}
|
||
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{}
|
||
}
|
||
return resp.Data, 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]))
|
||
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
|
||
step := pickHistogramStep(span)
|
||
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 {
|
||
// metric API 的 ts 为秒级浮点数(如 1.79e+09),query 日志的才是纳秒字符串
|
||
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),
|
||
})
|
||
}
|
||
}
|
||
return buckets, 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 <= 24*time.Hour:
|
||
return "15m"
|
||
case span <= 7*24*time.Hour:
|
||
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,
|
||
}
|
||
}
|