opsx
f3733f0bfc
- 品牌: Go 模块/数据文件/二进制/构建部署资产全部 opsx→lokistack;前端 localStorage 命名空间迁移(读旧写新,含按用户 id 前缀迁移);修复 nginx upstream 与 Makefile 引用已改名服务的断裂 - i18n: react-i18next 落地,common + 11 个页面域命名空间全站中英(菜单/顶栏/登录/列表页/表单/确认弹窗/状态徽章);设置页语言即时切换;修复词典自注册导致的循环导入白屏 - 用户管理: 批量删除(POST /users/batch-delete,不删自己/最后管理员,部分失败不阻断)+ 行复选/全选/批量工具条 - 主题: 浅色侧栏(中性灰激活态)、四态模式(浅色/深色/跟随系统/定时切换,按时区判定)、顶栏太阳月亮二态、设置页紧凑分段选择器 - UI: 登录记住用户名、CardActionButton 统一五页卡片操作钮、shadcn Badge 替换手写徽章、语义色修正 - 清理: 移除主机 Web 终端(前后端)、k8s-helm/Helm 遗留代码与字段、README 重写精简 - 修复: 侧栏与日志告警 order 重复、路由标题 i18n key 未翻译、词典双逗号等
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"
|
||
|
||
"lokistack/models"
|
||
"lokistack/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,
|
||
}
|
||
}
|