481 lines
15 KiB
Go
481 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"
|
||
)
|
||
|
||
// 探索类 API（标签浏览 / 直方图）与 log_proxy.go 查询代理拆分 // Logan.Li
|
||
|
||
// ListLokiLabels GET /api/v1/logs/loki/labels?datasource_id=&start=&end=
|
||
func (h *LogProxyHandler) ListLokiLabels(c *gin.Context) {
|
||
dsID, _ := strconv.ParseUint(c.Query("datasource_id"), 10, 64)
|
||
if dsID == 0 {
|
||
c.JSON(400, model.Response{Code: 400, Message: "datasource_id 必填"})
|
||
return
|
||
}
|
||
ds, err := loadLogDS(h.ds, uint(dsID))
|
||
if err != nil || ds.Type != "loki" {
|
||
c.JSON(400, model.Response{Code: 400, Message: "Loki 数据源无效"})
|
||
return
|
||
}
|
||
start, end := parseLogTimeRange(c.Query("start"), c.Query("end"))
|
||
params := url.Values{}
|
||
params.Set("start", fmt.Sprintf("%d", start.UnixNano()))
|
||
params.Set("end", fmt.Sprintf("%d", end.UnixNano()))
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, 15*time.Second)
|
||
target := strings.TrimRight(ds.URL, "/") + "/loki/api/v1/labels?" + 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)})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
labels := extractStringSlice(raw["data"])
|
||
c.JSON(200, model.Response{Code: 200, Data: labels})
|
||
}
|
||
|
||
// ListLokiLabelValues GET /api/v1/logs/loki/labels/:name/values
|
||
func (h *LogProxyHandler) ListLokiLabelValues(c *gin.Context) {
|
||
name := c.Param("name")
|
||
dsID, _ := strconv.ParseUint(c.Query("datasource_id"), 10, 64)
|
||
if dsID == 0 || name == "" {
|
||
c.JSON(400, model.Response{Code: 400, Message: "datasource_id 与 label 名必填"})
|
||
return
|
||
}
|
||
ds, err := loadLogDS(h.ds, uint(dsID))
|
||
if err != nil || ds.Type != "loki" {
|
||
c.JSON(400, model.Response{Code: 400, Message: "Loki 数据源无效"})
|
||
return
|
||
}
|
||
start, end := parseLogTimeRange(c.Query("start"), c.Query("end"))
|
||
params := url.Values{}
|
||
params.Set("start", fmt.Sprintf("%d", start.UnixNano()))
|
||
params.Set("end", fmt.Sprintf("%d", end.UnixNano()))
|
||
if q := extractLokiStreamSelector(strings.TrimSpace(c.Query("query"))); q != "" {
|
||
params.Set("query", q)
|
||
}
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, 15*time.Second)
|
||
target := fmt.Sprintf("%s/loki/api/v1/label/%s/values?%s", strings.TrimRight(ds.URL, "/"), url.PathEscape(name), 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)})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
values := extractStringSlice(raw["data"])
|
||
c.JSON(200, model.Response{Code: 200, Data: values})
|
||
}
|
||
|
||
// LokiHistogram POST /api/v1/logs/loki/histogram — 服务端日志量直方图 // Logan.Li
|
||
func (h *LogProxyHandler) LokiHistogram(c *gin.Context) {
|
||
var req struct {
|
||
DatasourceID uint `json:"datasource_id" binding:"required"`
|
||
Query string `json:"query" binding:"required"`
|
||
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
|
||
}
|
||
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
|
||
}
|
||
start, end := parseLogTimeRange(req.Start, req.End)
|
||
step := pickHistogramStep(end.Sub(start))
|
||
metricQuery := fmt.Sprintf("sum(count_over_time(%s[%s]))", req.Query, step)
|
||
params := url.Values{}
|
||
params.Set("query", metricQuery)
|
||
params.Set("start", fmt.Sprintf("%d", start.UnixNano()))
|
||
params.Set("end", fmt.Sprintf("%d", end.UnixNano()))
|
||
params.Set("step", step)
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, 30*time.Second)
|
||
target := strings.TrimRight(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)})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
buckets := flattenLokiMetricSeries(raw)
|
||
c.JSON(200, model.Response{Code: 200, Data: gin.H{"buckets": buckets}})
|
||
}
|
||
|
||
// ESHistogram POST /api/v1/logs/elasticsearch/histogram — ES date_histogram 聚合 // Logan.Li
|
||
func (h *LogProxyHandler) ESHistogram(c *gin.Context) {
|
||
var req struct {
|
||
DatasourceID uint `json:"datasource_id" binding:"required"`
|
||
Index string `json:"index"`
|
||
Query string `json:"query"`
|
||
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
|
||
}
|
||
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
|
||
}
|
||
start, end := parseLogTimeRange(req.Start, req.End)
|
||
interval := pickHistogramStep(end.Sub(start))
|
||
searchBody := buildESHistogramBody(req.Query, start, end, interval)
|
||
payload, _ := json.Marshal(searchBody)
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, 30*time.Second)
|
||
target := fmt.Sprintf("%s/%s/_search", strings.TrimRight(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)})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
buckets := flattenESHistogram(raw)
|
||
c.JSON(200, model.Response{Code: 200, Data: gin.H{"buckets": buckets}})
|
||
}
|
||
|
||
func parseLogTimeRange(startStr, endStr string) (time.Time, time.Time) {
|
||
end := time.Now()
|
||
start := end.Add(-1 * time.Hour)
|
||
if endStr != "" {
|
||
if t, e := parseLogTime(endStr); e == nil {
|
||
end = t
|
||
}
|
||
}
|
||
if startStr != "" {
|
||
if t, e := parseLogTime(startStr); e == nil {
|
||
start = t
|
||
}
|
||
}
|
||
return start, end
|
||
}
|
||
|
||
func pickHistogramStep(d time.Duration) string {
|
||
switch {
|
||
case d <= 15*time.Minute:
|
||
return "30s"
|
||
case d <= time.Hour:
|
||
return "1m"
|
||
case d <= 6*time.Hour:
|
||
return "5m"
|
||
case d <= 24*time.Hour:
|
||
return "15m"
|
||
case d <= 7*24*time.Hour:
|
||
return "1h"
|
||
default:
|
||
return "6h"
|
||
}
|
||
}
|
||
|
||
func extractStringSlice(v interface{}) []string {
|
||
arr, _ := v.([]interface{})
|
||
out := make([]string, 0, len(arr))
|
||
for _, item := range arr {
|
||
out = append(out, fmt.Sprintf("%v", item))
|
||
}
|
||
return out
|
||
}
|
||
|
||
func flattenLokiMetricSeries(raw map[string]interface{}) []map[string]interface{} {
|
||
out := []map[string]interface{}{}
|
||
data, _ := raw["data"].(map[string]interface{})
|
||
result, _ := data["result"].([]interface{})
|
||
for _, item := range result {
|
||
series, _ := item.(map[string]interface{})
|
||
values, _ := series["values"].([]interface{})
|
||
for _, val := range values {
|
||
pair, _ := val.([]interface{})
|
||
if len(pair) < 2 {
|
||
continue
|
||
}
|
||
ts := formatLokiMetricTimestamp(pair[0])
|
||
cnt, _ := strconv.ParseFloat(fmt.Sprintf("%v", pair[1]), 64)
|
||
out = append(out, map[string]interface{}{
|
||
"timestamp": ts,
|
||
"count": int64(cnt),
|
||
})
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func buildESHistogramBody(query string, start, end time.Time, interval string) map[string]interface{} {
|
||
body := map[string]interface{}{
|
||
"size": 0,
|
||
"query": map[string]interface{}{
|
||
"bool": map[string]interface{}{
|
||
"must": []interface{}{
|
||
map[string]interface{}{"query_string": map[string]interface{}{"query": defaultESQuery(query)}},
|
||
},
|
||
},
|
||
},
|
||
"aggs": map[string]interface{}{
|
||
"log_volume": map[string]interface{}{
|
||
"date_histogram": map[string]interface{}{
|
||
"field": "@timestamp",
|
||
"fixed_interval": interval,
|
||
"min_doc_count": 0,
|
||
},
|
||
},
|
||
},
|
||
}
|
||
rangeCond := map[string]interface{}{
|
||
"bool": map[string]interface{}{
|
||
"should": []interface{}{
|
||
map[string]interface{}{"range": map[string]interface{}{"@timestamp": map[string]string{
|
||
"gte": start.UTC().Format(time.RFC3339Nano),
|
||
"lte": end.UTC().Format(time.RFC3339Nano),
|
||
}}},
|
||
map[string]interface{}{"range": map[string]interface{}{"timestamp": map[string]string{
|
||
"gte": start.UTC().Format(time.RFC3339Nano),
|
||
"lte": end.UTC().Format(time.RFC3339Nano),
|
||
}}},
|
||
},
|
||
"minimum_should_match": 1,
|
||
},
|
||
}
|
||
body["query"] = map[string]interface{}{
|
||
"bool": map[string]interface{}{
|
||
"must": []interface{}{body["query"], rangeCond},
|
||
},
|
||
}
|
||
return body
|
||
}
|
||
|
||
func flattenESHistogram(raw map[string]interface{}) []map[string]interface{} {
|
||
out := []map[string]interface{}{}
|
||
aggs, _ := raw["aggregations"].(map[string]interface{})
|
||
vol, _ := aggs["log_volume"].(map[string]interface{})
|
||
buckets, _ := vol["buckets"].([]interface{})
|
||
for _, b := range buckets {
|
||
bucket, _ := b.(map[string]interface{})
|
||
ts, _ := bucket["key_as_string"].(string)
|
||
if ts == "" {
|
||
if key, ok := bucket["key"].(float64); ok {
|
||
ts = time.UnixMilli(int64(key)).UTC().Format(time.RFC3339Nano)
|
||
}
|
||
}
|
||
cnt, _ := bucket["doc_count"].(float64)
|
||
out = append(out, map[string]interface{}{"timestamp": ts, "count": int64(cnt)})
|
||
}
|
||
return out
|
||
}
|
||
|
||
// ESMapping GET /api/v1/logs/elasticsearch/mapping?datasource_id=&index=
|
||
func (h *LogProxyHandler) ESMapping(c *gin.Context) {
|
||
dsID, _ := strconv.ParseUint(c.Query("datasource_id"), 10, 64)
|
||
index := strings.TrimSpace(c.Query("index"))
|
||
if dsID == 0 {
|
||
c.JSON(400, model.Response{Code: 400, Message: "datasource_id 必填"})
|
||
return
|
||
}
|
||
if index == "" {
|
||
index = "*"
|
||
}
|
||
allowed, denyMsg := logaccess.CanQuery(h.db, middleware.GetUserID(c), middleware.GetIsAdmin(c), logaccess.QueryContext{
|
||
Engine: "elasticsearch", DatasourceID: uint(dsID), Index: index, Query: "*",
|
||
})
|
||
if !allowed {
|
||
c.JSON(403, model.Response{Code: 403, Message: denyMsg})
|
||
return
|
||
}
|
||
ds, err := loadLogDS(h.ds, uint(dsID))
|
||
if err != nil || ds.Type != "elasticsearch" {
|
||
c.JSON(400, model.Response{Code: 400, Message: "Elasticsearch 数据源无效"})
|
||
return
|
||
}
|
||
client := newLogHTTPClient(ds.SkipTLSVerify, 20*time.Second)
|
||
target := fmt.Sprintf("%s/%s/_mapping", strings.TrimRight(ds.URL, "/"), strings.Trim(index, "/"))
|
||
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: "获取 mapping 失败: " + 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)})
|
||
return
|
||
}
|
||
var raw map[string]interface{}
|
||
_ = json.Unmarshal(body, &raw)
|
||
fields := flattenESMapping(raw)
|
||
c.JSON(200, model.Response{Code: 200, Data: fields})
|
||
}
|
||
|
||
func flattenESMapping(raw map[string]interface{}) []map[string]string {
|
||
out := []map[string]string{}
|
||
seen := map[string]bool{}
|
||
var walk func(prefix string, props map[string]interface{})
|
||
walk = func(prefix string, props map[string]interface{}) {
|
||
for name, node := range props {
|
||
full := name
|
||
if prefix != "" {
|
||
full = prefix + "." + name
|
||
}
|
||
m, _ := node.(map[string]interface{})
|
||
if m == nil {
|
||
continue
|
||
}
|
||
if sub, ok := m["properties"].(map[string]interface{}); ok {
|
||
walk(full, sub)
|
||
continue
|
||
}
|
||
if fields, ok := m["fields"].(map[string]interface{}); ok {
|
||
walk(full, fields)
|
||
}
|
||
typ, _ := m["type"].(string)
|
||
if typ == "" {
|
||
typ = "object"
|
||
}
|
||
if !seen[full] {
|
||
seen[full] = true
|
||
out = append(out, map[string]string{"name": full, "type": mapESFieldType(typ)})
|
||
}
|
||
}
|
||
}
|
||
for _, idxDef := range raw {
|
||
idxMap, _ := idxDef.(map[string]interface{})
|
||
mappings, _ := idxMap["mappings"].(map[string]interface{})
|
||
props, _ := mappings["properties"].(map[string]interface{})
|
||
if props != nil {
|
||
walk("", props)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func mapESFieldType(esType string) string {
|
||
switch esType {
|
||
case "keyword", "text", "match_only_text":
|
||
return "keyword"
|
||
case "long", "integer", "short", "byte", "double", "float", "half_float", "scaled_float":
|
||
return "long"
|
||
case "date", "date_nanos":
|
||
return "date"
|
||
case "boolean":
|
||
return "boolean"
|
||
case "ip":
|
||
return "ip"
|
||
default:
|
||
return "text"
|
||
}
|
||
}
|
||
|
||
// extractLokiStreamSelector 从 LogQL 提取 {...} 选择器（Label API 仅接受 stream selector）// Logan.Li
|
||
func extractLokiStreamSelector(query string) string {
|
||
query = strings.TrimSpace(query)
|
||
start := strings.Index(query, "{")
|
||
if start < 0 {
|
||
return ""
|
||
}
|
||
depth := 0
|
||
for i := start; i < len(query); i++ {
|
||
switch query[i] {
|
||
case '{':
|
||
depth++
|
||
case '}':
|
||
depth--
|
||
if depth == 0 {
|
||
return query[start : i+1]
|
||
}
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// formatLokiMetricTimestamp 解析 Loki metric 序列时间戳（秒级 float / 科学计数法）// Logan.Li
|
||
func formatLokiMetricTimestamp(v interface{}) string {
|
||
switch t := v.(type) {
|
||
case float64:
|
||
sec := int64(t)
|
||
nsec := int64((t - float64(sec)) * 1e9)
|
||
return time.Unix(sec, nsec).UTC().Format(time.RFC3339)
|
||
case json.Number:
|
||
if f, err := t.Float64(); err == nil {
|
||
sec := int64(f)
|
||
return time.Unix(sec, 0).UTC().Format(time.RFC3339)
|
||
}
|
||
}
|
||
return formatLokiTimestamp(fmt.Sprintf("%v", v))
|
||
}
|