feat: 新增日志中心、功能优化与代码重构
- 新增 log_datasource.go / log_proxy.go / user_api_token.go 后端控制器 - 新增 log_datasource.go model - 新增 monitor/log-center.ts API 模块 - 新增 LogCenter 前端组件及页面 - 重构 deployment 相关组件 (HPDA/UpdateImageModal/bundle-yaml 等) - 新增 ResetPasswordDialog / code-snippet-block / copy-button 等 UI 组件 - 重构多个 k8s 相关组件 (TopologyView/ResourceVisualEditor/PodFileBrowser 等) - 更新多个 API 和工具函数 (auth.ts/monitor/index.ts/request.ts 等) - 更新静态资源 (index.html + 新 JS/CSS bundle) - 删除废弃静态资源文件
This commit is contained in:
Logan committed
1 parent
5ca0e6d83a
commit
8d46e8f352
197 files changed
+5014
-815
No files matched your search
@@ -0,0 +1,288 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"ezops/model"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// LogProxyHandler Loki / ES 无状态查询代理
|
||||
// 实时转发请求并解析展示字段，不将日志写入 AIOps 数据库 // Logan.Li
|
||||
type LogProxyHandler struct {
|
||||
ds *LogDSHandler
|
||||
}
|
||||
|
||||
const maxLogQueryLimit = 500 // 单次查询上限，防止平台承载过大数据 // Logan.Li
|
||||
|
||||
func NewLogProxyHandler(ds *LogDSHandler) *LogProxyHandler {
|
||||
return &LogProxyHandler{ds: ds}
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(400, model.Response{Code: 400, Message: err.Error()})
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
applyLogDSAuth(httpReq, ds.BasicAuth, ds.BasicAuthUser, ds.BasicAuthPass)
|
||||
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(502, model.Response{Code: 502, Message: "Loki 错误: " + string(body)})
|
||||
return
|
||||
}
|
||||
var raw map[string]interface{}
|
||||
_ = json.Unmarshal(body, &raw)
|
||||
entries := flattenLokiStreams(raw)
|
||||
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
||||
"entries": entries, "total": len(entries),
|
||||
}})
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(400, model.Response{Code: 400, Message: err.Error()})
|
||||
return
|
||||
}
|
||||
ds, err := loadLogDS(h.ds, req.DatasourceID)
|
||||
if err != nil || ds.Type != "elasticsearch" {
|
||||
c.JSON(400, model.Response{Code: 400, Message: "Elasticsearch 数据源无效"})
|
||||
return
|
||||
}
|
||||
index := req.Index
|
||||
if index == "" {
|
||||
index = ds.DefaultIndex
|
||||
}
|
||||
if index == "" {
|
||||
index = "*"
|
||||
}
|
||||
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,
|
||||
"sort": []map[string]interface{}{{"@timestamp": 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 != "" {
|
||||
rangeQ := map[string]interface{}{"range": map[string]interface{}{"@timestamp": map[string]string{}}}
|
||||
r := rangeQ["range"].(map[string]interface{})["@timestamp"].(map[string]string)
|
||||
if req.Start != "" {
|
||||
r["gte"] = req.Start
|
||||
}
|
||||
if req.End != "" {
|
||||
r["lte"] = req.End
|
||||
}
|
||||
searchBody["query"] = map[string]interface{}{
|
||||
"bool": map[string]interface{}{
|
||||
"must": []interface{}{
|
||||
searchBody["query"],
|
||||
rangeQ,
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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")
|
||||
applyLogDSAuth(httpReq, ds.BasicAuth, ds.BasicAuthUser, ds.BasicAuthPass)
|
||||
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(502, model.Response{Code: 502, Message: "ES 错误: " + string(body)})
|
||||
return
|
||||
}
|
||||
var raw map[string]interface{}
|
||||
_ = json.Unmarshal(body, &raw)
|
||||
entries := flattenESEntries(raw)
|
||||
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
||||
"entries": entries, "total": len(entries),
|
||||
}})
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
return time.Parse(time.RFC3339, 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 := 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 {
|
||||
out := []logEntryRow{}
|
||||
hitsWrap, _ := raw["hits"].(map[string]interface{})
|
||||
hits, _ := hitsWrap["hits"].([]interface{})
|
||||
for _, h := range hits {
|
||||
doc, _ := h.(map[string]interface{})
|
||||
src, _ := doc["_source"].(map[string]interface{})
|
||||
if src == nil {
|
||||
continue
|
||||
}
|
||||
msg := pickESMessage(src)
|
||||
level, _ := src["level"].(string)
|
||||
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,
|
||||
Raw: string(rawBytes),
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func pickESMessage(src map[string]interface{}) string {
|
||||
for _, key := range []string{"message", "msg", "log", "content"} {
|
||||
if v, ok := src[key]; ok {
|
||||
return fmt.Sprintf("%v", v)
|
||||
}
|
||||
}
|
||||
b, _ := json.Marshal(src)
|
||||
if len(b) > 500 {
|
||||
return string(b[:500]) + "..."
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
Reference in new issue
Block a user