Logan
1acb0c49af
- Agent CLI: 多轮对话、SSH 长连接、流式推理、Evolver 按用户隔离、历史侧栏 - 终端审计: 回放播放器、命令书签、筛选分页导出、命令输出统计、录制落盘修复 - 修复 session_commands 表迁移、Kill 竞态导致录制丢失、无录制会话回放提示 - 审计历史页只读（移除删除），移除回放弹框录制搜索
371 lines
11 KiB
Go
371 lines
11 KiB
Go
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 = 1000 // 单次查询返回日志行数硬上限（Loki limit / ES size 共用）// 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)
|
||
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)
|
||
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,
|
||
// ES 7+ 默认 track_total_hits 为 10000，必须显式开启才能返回真实命中总数 // Logan.Li
|
||
"track_total_hits": true,
|
||
"sort": []interface{}{
|
||
map[string]interface{}{"_score": 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,
|
||
},
|
||
},
|
||
}
|
||
}
|
||
}
|
||
|
||
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 := flattenESEntries(raw)
|
||
total := esHitsTotal(raw)
|
||
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
||
"entries": entries, "total": total,
|
||
}})
|
||
}
|
||
|
||
// 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"
|
||
}
|
||
}
|
||
|
||
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 := 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
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
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)
|
||
}
|