310 lines
8.5 KiB
Go
310 lines
8.5 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 = 500
|
|
|
|
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": []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 格式要求 +00:00 而非 Z；兼容 @timestamp 与 timestamp // Logan.Li
|
|
buildRangeVal := func(v string) string {
|
|
return strings.Replace(v, "Z", "+00:00", 1)
|
|
}
|
|
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"] = buildRangeVal(req.Start)
|
|
}
|
|
if req.End != "" {
|
|
r["lte"] = buildRangeVal(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")
|
|
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
|
|
}
|
|
// ES 7.2 date_time 不接受 Z，转 +00:00
|
|
s = strings.Replace(s, "Z", "+00:00", 1)
|
|
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)
|
|
}
|