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)) }