aiops/backend/controller/log_proxy.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 仅接受 .SSSZ 格式；兼容 @timestamp 与 timestamp // Logan.Li
			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"] = req.Start
					}
					if req.End != "" {
						r["lte"] = 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
	}
	// 解析 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
}

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