Logan
7e1c8f583c
- 工具配置：新增卸载 API/UI、Go 工具链、流水线构建自动安装依赖 - 修复 ListTools VerifyTool 递归栈溢出导致获取工具配置失败 - Node.js Alpine 安装改用 unofficial-builds musl 预编译包与 index.json 解析 LTS - 站点拨测：详情页路由、告警自动恢复、并发拨测与默认可观测频率 - 日志中心：RBAC、查询构建器、ES/Loki 增强与审计 request/response 展示 - 发布配置：构建节点、流水线远程执行、Deploy RBAC 导航拆分
531 lines
15 KiB
Go
531 lines
15 KiB
Go
package controller
|
||
|
||
import (
|
||
"bytes"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"net/url"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"ezops/middleware"
|
||
"ezops/model"
|
||
"ezops/pkg/logaccess"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// LogProxyHandler Loki / ES 无状态查询代理
|
||
// 实时转发请求并解析展示字段，不将日志写入 AIOps 数据库 // Logan.Li
|
||
type LogProxyHandler struct {
|
||
ds *LogDSHandler
|
||
db *gorm.DB
|
||
}
|
||
|
||
const maxLogQueryLimit = 1000 // 单次查询返回日志行数硬上限（Loki limit / ES size 共用）// Logan.Li
|
||
|
||
func NewLogProxyHandler(ds *LogDSHandler, db *gorm.DB) *LogProxyHandler {
|
||
return &LogProxyHandler{ds: ds, db: db}
|
||
}
|
||
|
||
// QueryLoki POST /api/v1/logs/loki/query
|
||
func (h *LogProxyHandler) QueryLoki(c *gin.Context) {
|
||
var req struct {
|
||
DatasourceID uint `json:"datasource_id" binding:"required"`
|
||
Query string `json:"query" binding:"required"`
|
||
Limit int `json:"limit"`
|
||
Start string `json:"start"`
|
||
End string `json:"end"`
|
||
Direction string `json:"direction"`
|
||
Cursor string `json:"cursor"` // 纳秒时间戳，向后翻页时作为 end // Logan.Li
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
c.JSON(400, model.Response{Code: 400, Message: err.Error()})
|
||
return
|
||
}
|
||
allowed, denyMsg := logaccess.CanQuery(h.db, middleware.GetUserID(c), middleware.GetIsAdmin(c), logaccess.QueryContext{
|
||
Engine: "loki", DatasourceID: req.DatasourceID, Query: req.Query,
|
||
})
|
||
if !allowed {
|
||
c.JSON(403, model.Response{Code: 403, Message: denyMsg})
|
||
return
|
||
}
|
||
ds, err := loadLogDS(h.ds, req.DatasourceID)
|
||
if err != nil || ds.Type != "loki" {
|
||
c.JSON(400, model.Response{Code: 400, Message: "Loki 数据源无效"})
|
||
return
|
||
}
|
||
if req.Limit <= 0 {
|
||
req.Limit = 200
|
||
}
|
||
if req.Limit > maxLogQueryLimit {
|
||
req.Limit = maxLogQueryLimit
|
||
}
|
||
if req.Direction == "" {
|
||
req.Direction = "backward"
|
||
}
|
||
end := time.Now()
|
||
start := end.Add(-1 * time.Hour)
|
||
if req.End != "" {
|
||
if t, e := parseLogTime(req.End); e == nil {
|
||
end = t
|
||
}
|
||
}
|
||
if req.Start != "" {
|
||
if t, e := parseLogTime(req.Start); e == nil {
|
||
start = t
|
||
}
|
||
}
|
||
if req.Cursor != "" {
|
||
if n, err := strconv.ParseInt(strings.TrimSpace(req.Cursor), 10, 64); err == nil && n > 0 {
|
||
end = time.Unix(0, n)
|
||
}
|
||
}
|
||
|
||
params := url.Values{}
|
||
params.Set("query", req.Query)
|
||
params.Set("limit", strconv.Itoa(req.Limit))
|
||
params.Set("direction", req.Direction)
|
||
params.Set("start", fmt.Sprintf("%d", start.UnixNano()))
|
||
params.Set("end", fmt.Sprintf("%d", end.UnixNano()))
|
||
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, time.Duration(60)*time.Second)
|
||
target := ds.URL + "/loki/api/v1/query_range?" + params.Encode()
|
||
httpReq, _ := http.NewRequest(http.MethodGet, target, nil)
|
||
applyLogDSAuthV2(httpReq, ds)
|
||
resp, err := client.Do(httpReq)
|
||
if err != nil {
|
||
c.JSON(502, model.Response{Code: 502, Message: "查询失败: " + err.Error()})
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
body, _ := io.ReadAll(resp.Body)
|
||
if resp.StatusCode >= 400 {
|
||
c.JSON(logProxyErrStatus(resp.StatusCode), model.Response{Code: logProxyErrStatus(resp.StatusCode), Message: "Loki 错误: " + string(body), Data: map[string]string{"category": logProxyErrCategory(resp.StatusCode)}})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
entries := flattenLokiStreams(raw)
|
||
nextCursor := ""
|
||
if len(entries) > 0 && len(entries) >= req.Limit {
|
||
nextCursor = lokiEntryCursor(entries[len(entries)-1].Timestamp)
|
||
}
|
||
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
||
"entries": entries, "total": len(entries), "has_more": nextCursor != "", "next_cursor": nextCursor,
|
||
}})
|
||
}
|
||
|
||
// QueryES POST /api/v1/logs/elasticsearch/query
|
||
func (h *LogProxyHandler) QueryES(c *gin.Context) {
|
||
var req struct {
|
||
DatasourceID uint `json:"datasource_id" binding:"required"`
|
||
Index string `json:"index"`
|
||
Query string `json:"query"`
|
||
QueryBody map[string]interface{} `json:"query_body"`
|
||
Size int `json:"size"`
|
||
Start string `json:"start"`
|
||
End string `json:"end"`
|
||
SearchAfter []interface{} `json:"search_after"` // ES 深分页游标 // Logan.Li
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
c.JSON(400, model.Response{Code: 400, Message: err.Error()})
|
||
return
|
||
}
|
||
index := req.Index
|
||
if index == "" {
|
||
if dsPreview, err := loadLogDS(h.ds, req.DatasourceID); err == nil {
|
||
index = dsPreview.DefaultIndex
|
||
}
|
||
}
|
||
if index == "" {
|
||
index = "*"
|
||
}
|
||
allowed, denyMsg := logaccess.CanQuery(h.db, middleware.GetUserID(c), middleware.GetIsAdmin(c), logaccess.QueryContext{
|
||
Engine: "elasticsearch", DatasourceID: req.DatasourceID, Index: index, Query: req.Query,
|
||
})
|
||
if !allowed {
|
||
c.JSON(403, model.Response{Code: 403, Message: denyMsg})
|
||
return
|
||
}
|
||
ds, err := loadLogDS(h.ds, req.DatasourceID)
|
||
if err != nil || ds.Type != "elasticsearch" {
|
||
c.JSON(400, model.Response{Code: 400, Message: "Elasticsearch 数据源无效"})
|
||
return
|
||
}
|
||
if req.Size <= 0 {
|
||
req.Size = 200
|
||
}
|
||
if req.Size > maxLogQueryLimit {
|
||
req.Size = maxLogQueryLimit
|
||
}
|
||
|
||
searchBody := req.QueryBody
|
||
if searchBody == nil {
|
||
searchBody = map[string]interface{}{
|
||
"size": req.Size,
|
||
// ES 7+ 默认 track_total_hits 为 10000，必须显式开启才能返回真实命中总数 // Logan.Li
|
||
"track_total_hits": true,
|
||
"sort": []interface{}{
|
||
map[string]interface{}{"@timestamp": map[string]string{"order": "desc", "unmapped_type": "date"}},
|
||
map[string]interface{}{"_id": map[string]string{"order": "desc"}},
|
||
},
|
||
"query": map[string]interface{}{
|
||
"bool": map[string]interface{}{
|
||
"must": []interface{}{
|
||
map[string]interface{}{"query_string": map[string]interface{}{"query": defaultESQuery(req.Query)}},
|
||
},
|
||
},
|
||
},
|
||
}
|
||
if req.Start != "" || req.End != "" {
|
||
// ES 7.2 date_time 仅接受 .SSSZ 格式（strict_date_optional_time 不含 epoch_millis）
|
||
// 兼容各种前端时间格式：补全毫秒 // Logan.Li
|
||
normalizeTime := func(s string) string {
|
||
if s == "" {
|
||
return s
|
||
}
|
||
// 空格 → T（兼容前端未正确转换的格式）// Logan.Li
|
||
s = strings.Replace(s, " ", "T", 1)
|
||
// 查找时间部分结束位置（Z 或 +）
|
||
tzIdx := strings.IndexAny(s, "Z+")
|
||
timePart := s
|
||
tzPart := ""
|
||
if tzIdx >= 0 {
|
||
timePart = s[:tzIdx]
|
||
tzPart = s[tzIdx:]
|
||
}
|
||
// 如果已有毫秒部分（.后跟数字），直接返回
|
||
if dotIdx := strings.LastIndexByte(timePart, '.'); dotIdx >= 0 {
|
||
if dotIdx+1 < len(timePart) && timePart[dotIdx+1] >= '0' && timePart[dotIdx+1] <= '9' {
|
||
return s
|
||
}
|
||
}
|
||
return timePart + ".000" + tzPart
|
||
}
|
||
rangeCond := map[string]interface{}{
|
||
"bool": map[string]interface{}{
|
||
"should": []interface{}{
|
||
map[string]interface{}{"range": map[string]interface{}{"@timestamp": map[string]string{}}},
|
||
map[string]interface{}{"range": map[string]interface{}{"timestamp": map[string]string{}}},
|
||
},
|
||
"minimum_should_match": 1,
|
||
},
|
||
}
|
||
for _, clause := range rangeCond["bool"].(map[string]interface{})["should"].([]interface{}) {
|
||
rng := clause.(map[string]interface{})["range"].(map[string]interface{})
|
||
for _, v := range rng {
|
||
r := v.(map[string]string)
|
||
if req.Start != "" {
|
||
r["gte"] = normalizeTime(req.Start)
|
||
}
|
||
if req.End != "" {
|
||
r["lte"] = normalizeTime(req.End)
|
||
}
|
||
}
|
||
}
|
||
searchBody["query"] = map[string]interface{}{
|
||
"bool": map[string]interface{}{
|
||
"must": []interface{}{
|
||
searchBody["query"],
|
||
rangeCond,
|
||
},
|
||
},
|
||
}
|
||
}
|
||
}
|
||
if len(req.SearchAfter) > 0 {
|
||
searchBody["search_after"] = req.SearchAfter
|
||
}
|
||
|
||
payload, _ := json.Marshal(searchBody)
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, 60*time.Second)
|
||
target := fmt.Sprintf("%s/%s/_search", ds.URL, strings.Trim(index, "/"))
|
||
httpReq, _ := http.NewRequest(http.MethodPost, target, bytes.NewReader(payload))
|
||
httpReq.Header.Set("Content-Type", "application/json")
|
||
applyLogDSAuthV2(httpReq, ds)
|
||
resp, err := client.Do(httpReq)
|
||
if err != nil {
|
||
c.JSON(502, model.Response{Code: 502, Message: "查询失败: " + err.Error()})
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
body, _ := io.ReadAll(resp.Body)
|
||
if resp.StatusCode >= 400 {
|
||
c.JSON(logProxyErrStatus(resp.StatusCode), model.Response{Code: logProxyErrStatus(resp.StatusCode), Message: "ES 错误: " + string(body), Data: map[string]string{"category": logProxyErrCategory(resp.StatusCode)}})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
entries, nextSearchAfter := flattenESEntries(raw)
|
||
total := esHitsTotal(raw)
|
||
hasMore := len(entries) >= req.Size && len(nextSearchAfter) > 0
|
||
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
||
"entries": entries, "total": total, "has_more": hasMore, "search_after": nextSearchAfter,
|
||
}})
|
||
}
|
||
|
||
// logProxyErrStatus 把下游 HTTP 状态码映射为合适的 API 状态码 // Logan.Li
|
||
func logProxyErrStatus(status int) int {
|
||
if status >= 400 && status < 500 {
|
||
return 400
|
||
}
|
||
return 502
|
||
}
|
||
|
||
// logProxyErrCategory 对下游错误进行分类，便于前端展示中文提示 // Logan.Li
|
||
func logProxyErrCategory(status int) string {
|
||
switch status {
|
||
case 401:
|
||
return "auth_failed"
|
||
case 403:
|
||
return "forbidden"
|
||
case 400, 404:
|
||
return "bad_request"
|
||
default:
|
||
if status >= 500 {
|
||
return "server_error"
|
||
}
|
||
return "unknown"
|
||
}
|
||
}
|
||
|
||
// formatLokiTimestamp 将 Loki 纳秒/秒级（含科学计数法）时间戳转为 RFC3339 // Logan.Li
|
||
func formatLokiTimestamp(ts string) string {
|
||
ts = strings.TrimSpace(ts)
|
||
if f, err := strconv.ParseFloat(ts, 64); err == nil {
|
||
if strings.Contains(ts, ".") || strings.Contains(strings.ToLower(ts), "e") {
|
||
sec := int64(f)
|
||
nsec := int64((f - float64(sec)) * 1e9)
|
||
return time.Unix(sec, nsec).UTC().Format(time.RFC3339)
|
||
}
|
||
}
|
||
if n, err := strconv.ParseInt(ts, 10, 64); err == nil {
|
||
if n > 1e15 {
|
||
return time.Unix(0, n).UTC().Format(time.RFC3339Nano)
|
||
}
|
||
if n > 1e12 {
|
||
return time.UnixMilli(n).UTC().Format(time.RFC3339Nano)
|
||
}
|
||
return time.Unix(n, 0).UTC().Format(time.RFC3339)
|
||
}
|
||
return ts
|
||
}
|
||
|
||
func parseLogTime(s string) (time.Time, error) {
|
||
if n, err := strconv.ParseInt(s, 10, 64); err == nil {
|
||
if n > 1e15 {
|
||
return time.Unix(0, n), nil
|
||
}
|
||
return time.Unix(n, 0), nil
|
||
}
|
||
// 解析 ISO8601（含 .SSSZ 毫秒后缀，如 2026-06-20T09:33:17.000Z）
|
||
return time.Parse(time.RFC3339Nano, s)
|
||
}
|
||
|
||
func defaultESQuery(q string) string {
|
||
if strings.TrimSpace(q) == "" {
|
||
return "*"
|
||
}
|
||
return q
|
||
}
|
||
|
||
type logEntryRow struct {
|
||
Timestamp string `json:"timestamp"`
|
||
Level string `json:"level,omitempty"`
|
||
Message string `json:"message"`
|
||
Labels map[string]string `json:"labels,omitempty"`
|
||
Raw string `json:"raw,omitempty"`
|
||
}
|
||
|
||
func flattenLokiStreams(raw map[string]interface{}) []logEntryRow {
|
||
out := []logEntryRow{}
|
||
data, _ := raw["data"].(map[string]interface{})
|
||
result, _ := data["result"].([]interface{})
|
||
for _, item := range result {
|
||
stream, _ := item.(map[string]interface{})
|
||
labels, _ := stream["stream"].(map[string]interface{})
|
||
labelMap := map[string]string{}
|
||
for k, v := range labels {
|
||
labelMap[k] = fmt.Sprintf("%v", v)
|
||
}
|
||
values, _ := stream["values"].([]interface{})
|
||
for _, val := range values {
|
||
pair, _ := val.([]interface{})
|
||
if len(pair) < 2 {
|
||
continue
|
||
}
|
||
ts := formatLokiTimestamp(fmt.Sprintf("%v", pair[0]))
|
||
msg := fmt.Sprintf("%v", pair[1])
|
||
out = append(out, logEntryRow{
|
||
Timestamp: ts,
|
||
Level: labelMap["level"],
|
||
Message: msg,
|
||
Labels: labelMap,
|
||
Raw: msg,
|
||
})
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func flattenESEntries(raw map[string]interface{}) ([]logEntryRow, []interface{}) {
|
||
out := []logEntryRow{}
|
||
var nextAfter []interface{}
|
||
hitsWrap, _ := raw["hits"].(map[string]interface{})
|
||
hits, _ := hitsWrap["hits"].([]interface{})
|
||
for i, h := range hits {
|
||
doc, _ := h.(map[string]interface{})
|
||
src, _ := doc["_source"].(map[string]interface{})
|
||
if src == nil {
|
||
continue
|
||
}
|
||
msg := pickESMessage(src)
|
||
level := pickESLevel(src)
|
||
ts, _ := src["@timestamp"].(string)
|
||
if ts == "" {
|
||
ts = fmt.Sprintf("%v", src["timestamp"])
|
||
}
|
||
rawBytes, _ := json.Marshal(src)
|
||
out = append(out, logEntryRow{
|
||
Timestamp: ts,
|
||
Level: level,
|
||
Message: msg,
|
||
Labels: extractESLabels(src),
|
||
Raw: string(rawBytes),
|
||
})
|
||
if i == len(hits)-1 {
|
||
if sortVals, ok := doc["sort"].([]interface{}); ok {
|
||
nextAfter = sortVals
|
||
}
|
||
}
|
||
}
|
||
return out, nextAfter
|
||
}
|
||
|
||
// lokiEntryCursor 从 RFC3339 时间戳反推纳秒游标 // Logan.Li
|
||
func lokiEntryCursor(ts string) string {
|
||
if t, err := time.Parse(time.RFC3339Nano, ts); err == nil {
|
||
return strconv.FormatInt(t.UnixNano(), 10)
|
||
}
|
||
if t, err := time.Parse(time.RFC3339, ts); err == nil {
|
||
return strconv.FormatInt(t.UnixNano(), 10)
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// extractESLabels 从 _source 提取常用字段作为 labels（便于字段过滤）// Logan.Li
|
||
func extractESLabels(src map[string]interface{}) map[string]string {
|
||
labels := map[string]string{}
|
||
for _, key := range []string{"app", "service", "service_name", "host.name", "kubernetes.namespace", "namespace", "container.name", "level"} {
|
||
if v := pickESField(src, key); v != "" {
|
||
labels[key] = v
|
||
}
|
||
}
|
||
if lv := pickESLevel(src); lv != "" {
|
||
labels["level"] = lv
|
||
}
|
||
return labels
|
||
}
|
||
|
||
func pickESField(src map[string]interface{}, key string) string {
|
||
if strings.Contains(key, ".") {
|
||
parts := strings.Split(key, ".")
|
||
cur := interface{}(src)
|
||
for _, p := range parts {
|
||
m, ok := cur.(map[string]interface{})
|
||
if !ok {
|
||
return ""
|
||
}
|
||
cur, ok = m[p]
|
||
if !ok {
|
||
return ""
|
||
}
|
||
}
|
||
return stringifyESField(cur)
|
||
}
|
||
if v, ok := src[key]; ok {
|
||
return stringifyESField(v)
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// esHitsTotal 从 ES 响应中提取真实命中总数（兼容 ES 7+ total.value 与旧版 total）// Logan.Li
|
||
func esHitsTotal(raw map[string]interface{}) int64 {
|
||
hitsWrap, _ := raw["hits"].(map[string]interface{})
|
||
if total, ok := hitsWrap["total"].(map[string]interface{}); ok {
|
||
if v, ok := total["value"].(float64); ok {
|
||
return int64(v)
|
||
}
|
||
}
|
||
if v, ok := hitsWrap["total"].(float64); ok {
|
||
return int64(v)
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// pickESLevel 提取日志级别（兼容顶层 level 与 ECS/Filebeat 的 log.level） // Logan.Li
|
||
func pickESLevel(src map[string]interface{}) string {
|
||
if v, ok := src["level"].(string); ok && v != "" {
|
||
return v
|
||
}
|
||
if v, ok := src["log.level"].(string); ok && v != "" {
|
||
return v
|
||
}
|
||
if logObj, ok := src["log"].(map[string]interface{}); ok {
|
||
if v, ok := logObj["level"].(string); ok && v != "" {
|
||
return v
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// stringifyESField 将 ES 字段值转为展示字符串 // Logan.Li
|
||
func stringifyESField(v interface{}) string {
|
||
switch t := v.(type) {
|
||
case string:
|
||
return t
|
||
case float64:
|
||
return fmt.Sprintf("%v", t)
|
||
case bool:
|
||
return fmt.Sprintf("%t", t)
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// pickESMessage 提取日志正文（兼容 message / ECS log.original / event.original） // Logan.Li
|
||
func pickESMessage(src map[string]interface{}) string {
|
||
for _, key := range []string{"message", "msg", "content", "event.original"} {
|
||
if v, ok := src[key]; ok {
|
||
if s := stringifyESField(v); s != "" {
|
||
return s
|
||
}
|
||
}
|
||
}
|
||
if logObj, ok := src["log"].(map[string]interface{}); ok {
|
||
for _, key := range []string{"original", "message"} {
|
||
if v, ok := logObj[key]; ok {
|
||
if s := stringifyESField(v); s != "" {
|
||
return s
|
||
}
|
||
}
|
||
}
|
||
}
|
||
// 顶层 log 为字符串时（部分旧索引）
|
||
if v, ok := src["log"]; ok {
|
||
if s := stringifyESField(v); s != "" {
|
||
return s
|
||
}
|
||
}
|
||
b, _ := json.Marshal(src)
|
||
if len(b) > 500 {
|
||
return string(b[:500]) + "..."
|
||
}
|
||
return string(b)
|
||
}
|