541 lines
17 KiB
Go
541 lines
17 KiB
Go
package controller
|
|
|
|
import (
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"net/http"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"ezops/model"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
)
|
|
|
|
// TraceProxyHandler 链路追踪查询代理（Jaeger / Tempo 兼容） // Logan.Li
|
|
// 转发查询请求到外部 Jaeger/Tempo 服务，并将响应转换为前端期望的格式
|
|
// 无数据源时返回 demo 数据让 UI 保持可用
|
|
type TraceProxyHandler struct {
|
|
ds *TraceDSHandler
|
|
}
|
|
|
|
func NewTraceProxyHandler(ds *TraceDSHandler) *TraceProxyHandler {
|
|
return &TraceProxyHandler{ds: ds}
|
|
}
|
|
|
|
// ---- Jaeger 原始响应结构 ----
|
|
|
|
type jaegerServicesResp struct {
|
|
Data []string `json:"data"`
|
|
}
|
|
|
|
type jaegerTraceResp struct {
|
|
Data []jaegerTrace `json:"data"`
|
|
Total int `json:"total"`
|
|
}
|
|
|
|
type jaegerTrace struct {
|
|
TraceID string `json:"traceID"`
|
|
Spans []jaegerSpan `json:"spans"`
|
|
Processes map[string]jaegerProc `json:"processes"`
|
|
}
|
|
|
|
type jaegerSpan struct {
|
|
TraceID string `json:"traceID"`
|
|
SpanID string `json:"spanID"`
|
|
OperationName string `json:"operationName"`
|
|
References []jaegerRef `json:"references"`
|
|
StartTime int64 `json:"startTime"` // microseconds
|
|
Duration int64 `json:"duration"` // microseconds
|
|
Tags []jaegerTag `json:"tags"`
|
|
ProcessID string `json:"processID"`
|
|
Logs []jaegerLog `json:"logs"`
|
|
}
|
|
|
|
type jaegerRef struct {
|
|
RefType string `json:"refType"`
|
|
TraceID string `json:"traceID"`
|
|
SpanID string `json:"spanID"`
|
|
}
|
|
|
|
type jaegerProc struct {
|
|
ServiceName string `json:"serviceName"`
|
|
Tags []jaegerTag `json:"tags"`
|
|
}
|
|
|
|
type jaegerTag struct {
|
|
Key string `json:"key"`
|
|
Type string `json:"type"`
|
|
Value interface{} `json:"value"`
|
|
}
|
|
|
|
type jaegerLog struct {
|
|
Timestamp int64 `json:"timestamp"`
|
|
Fields []jaegerTag `json:"fields"`
|
|
}
|
|
|
|
// ---- 前端期望的格式 ----
|
|
|
|
type traceListItem struct {
|
|
ID string `json:"id"`
|
|
TraceID string `json:"traceId"`
|
|
ServiceName string `json:"serviceName"`
|
|
OperationName string `json:"operationName"`
|
|
StartTime string `json:"startTime"`
|
|
Duration float64 `json:"duration"` // ms
|
|
Status string `json:"status"`
|
|
SpanCount int `json:"spanCount"`
|
|
Tags map[string]string `json:"tags"`
|
|
}
|
|
|
|
type spanItem struct {
|
|
ID string `json:"id"`
|
|
SpanID string `json:"spanId"`
|
|
ParentSpanID string `json:"parentSpanId,omitempty"`
|
|
ServiceName string `json:"serviceName"`
|
|
OperationName string `json:"operationName"`
|
|
StartTime string `json:"startTime"`
|
|
Duration float64 `json:"duration"` // ms
|
|
Status string `json:"status"`
|
|
Kind string `json:"kind"`
|
|
Tags map[string]string `json:"tags"`
|
|
Logs []spanLogItem `json:"logs"`
|
|
}
|
|
|
|
type spanLogItem struct {
|
|
Timestamp string `json:"timestamp"`
|
|
Fields map[string]string `json:"fields"`
|
|
}
|
|
|
|
type serviceMetric struct {
|
|
ServiceName string `json:"serviceName"`
|
|
RequestCount int `json:"requestCount"`
|
|
ErrorRate float64 `json:"errorRate"`
|
|
AvgDuration float64 `json:"avgDuration"` // ms
|
|
P95Duration float64 `json:"p95Duration"`
|
|
P99Duration float64 `json:"p99Duration"`
|
|
}
|
|
|
|
// ---- HTTP 辅助 ----
|
|
|
|
func newTraceHTTPClient(skipTLS bool, timeout time.Duration) *http.Client {
|
|
tr := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: skipTLS}}
|
|
return &http.Client{Transport: tr, Timeout: timeout}
|
|
}
|
|
|
|
func (h *TraceProxyHandler) doGet(ds *model.TraceDataSource, path string) ([]byte, int, error) {
|
|
target := ds.URL + path
|
|
client := newTraceHTTPClient(ds.SkipTLSVerify, 30*time.Second)
|
|
req, _ := http.NewRequest("GET", target, nil)
|
|
if ds.BasicAuth && ds.BasicAuthUser != "" {
|
|
req.SetBasicAuth(ds.BasicAuthUser, ds.BasicAuthPass)
|
|
}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer resp.Body.Close()
|
|
body, _ := io.ReadAll(resp.Body)
|
|
return body, resp.StatusCode, nil
|
|
}
|
|
|
|
// ---- 转换函数 ----
|
|
|
|
func tagsToMap(tags []jaegerTag) map[string]string {
|
|
m := map[string]string{}
|
|
for _, t := range tags {
|
|
m[t.Key] = fmt.Sprintf("%v", t.Value)
|
|
}
|
|
return m
|
|
}
|
|
|
|
func spanStatus(spans []jaegerSpan) string {
|
|
for _, s := range spans {
|
|
for _, t := range s.Tags {
|
|
if t.Key == "error" {
|
|
if b, ok := t.Value.(bool); ok && b {
|
|
return "error"
|
|
}
|
|
}
|
|
if t.Key == "http.status_code" {
|
|
if code, ok := toInt(t.Value); ok && code >= 400 {
|
|
return "error"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return "success"
|
|
}
|
|
|
|
func toInt(v interface{}) (int, bool) {
|
|
switch n := v.(type) {
|
|
case float64:
|
|
return int(n), true
|
|
case int:
|
|
return n, true
|
|
case int64:
|
|
return int(n), true
|
|
}
|
|
return 0, false
|
|
}
|
|
|
|
func formatTraceTime(us int64) string {
|
|
return time.UnixMicro(us).Format("2006-01-02 15:04:05")
|
|
}
|
|
|
|
func kindFromTags(tags []jaegerTag) string {
|
|
for _, t := range tags {
|
|
if t.Key == "span.kind" {
|
|
if s, ok := t.Value.(string); ok {
|
|
return s
|
|
}
|
|
}
|
|
}
|
|
return "internal"
|
|
}
|
|
|
|
// ---- API: GetServices GET /api/v1/traces/services ----
|
|
|
|
func (h *TraceProxyHandler) GetServices(c *gin.Context) {
|
|
timeRange := c.DefaultQuery("timeRange", "1h")
|
|
ds, err := h.ds.getDefaultTraceDS()
|
|
if err != nil || ds == nil {
|
|
// demo 数据
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
|
"data": demoServiceMetrics(),
|
|
"total": 5,
|
|
"demo": true,
|
|
}})
|
|
return
|
|
}
|
|
|
|
body, status, err := h.doGet(ds, "/api/services")
|
|
if err != nil || status != 200 {
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
|
"data": demoServiceMetrics(),
|
|
"total": 5,
|
|
"demo": true,
|
|
}})
|
|
return
|
|
}
|
|
|
|
var svcResp jaegerServicesResp
|
|
if json.Unmarshal(body, &svcResp) != nil {
|
|
c.JSON(500, model.Response{Code: 500, Message: "解析服务列表失败"})
|
|
return
|
|
}
|
|
|
|
// 查询 traces 来计算服务指标
|
|
services := buildServiceMetricsFromTraces(h, ds, svcResp.Data, timeRange)
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
|
"data": services,
|
|
"total": len(services),
|
|
}})
|
|
}
|
|
|
|
func buildServiceMetricsFromTraces(h *TraceProxyHandler, ds *model.TraceDataSource, serviceNames []string, timeRange string) []serviceMetric {
|
|
// 查询最近 traces 来聚合服务指标
|
|
dur := parseTraceRange(timeRange)
|
|
end := time.Now()
|
|
start := end.Add(-dur)
|
|
traces := h.queryTraces(ds, "", 200, start, end)
|
|
|
|
// 按服务聚合
|
|
type agg struct {
|
|
durations []float64
|
|
errors int
|
|
}
|
|
aggMap := map[string]*agg{}
|
|
for _, name := range serviceNames {
|
|
aggMap[name] = &agg{}
|
|
}
|
|
|
|
for _, t := range traces {
|
|
for _, sp := range t.Spans {
|
|
proc, ok := t.Processes[sp.ProcessID]
|
|
if !ok {
|
|
continue
|
|
}
|
|
svcName := proc.ServiceName
|
|
a, exists := aggMap[svcName]
|
|
if !exists {
|
|
a = &agg{}
|
|
aggMap[svcName] = a
|
|
}
|
|
a.durations = append(a.durations, float64(sp.Duration)/1000.0) // μs→ms
|
|
isErr := false
|
|
for _, tag := range sp.Tags {
|
|
if tag.Key == "error" {
|
|
if b, ok := tag.Value.(bool); ok && b {
|
|
isErr = true
|
|
}
|
|
}
|
|
}
|
|
if isErr {
|
|
a.errors++
|
|
}
|
|
}
|
|
}
|
|
|
|
result := []serviceMetric{}
|
|
for _, name := range serviceNames {
|
|
a := aggMap[name]
|
|
if len(a.durations) == 0 {
|
|
result = append(result, serviceMetric{
|
|
ServiceName: name, RequestCount: 0, ErrorRate: 0,
|
|
AvgDuration: 0, P95Duration: 0, P99Duration: 0,
|
|
})
|
|
continue
|
|
}
|
|
sort.Float64s(a.durations)
|
|
total := len(a.durations)
|
|
sum := 0.0
|
|
for _, d := range a.durations {
|
|
sum += d
|
|
}
|
|
avg := sum / float64(total)
|
|
p95 := percentile(a.durations, 0.95)
|
|
p99 := percentile(a.durations, 0.99)
|
|
errRate := float64(a.errors) / float64(total) * 100
|
|
result = append(result, serviceMetric{
|
|
ServiceName: name,
|
|
RequestCount: total,
|
|
ErrorRate: math.Round(errRate*100) / 100,
|
|
AvgDuration: math.Round(avg*100) / 100,
|
|
P95Duration: math.Round(p95*100) / 100,
|
|
P99Duration: math.Round(p99*100) / 100,
|
|
})
|
|
}
|
|
return result
|
|
}
|
|
|
|
func percentile(sorted []float64, p float64) float64 {
|
|
if len(sorted) == 0 {
|
|
return 0
|
|
}
|
|
idx := int(float64(len(sorted)-1) * p)
|
|
return sorted[idx]
|
|
}
|
|
|
|
func parseTraceRange(r string) time.Duration {
|
|
switch r {
|
|
case "15m":
|
|
return 15 * time.Minute
|
|
case "1h":
|
|
return time.Hour
|
|
case "6h":
|
|
return 6 * time.Hour
|
|
case "24h":
|
|
return 24 * time.Hour
|
|
case "7d":
|
|
return 7 * 24 * time.Hour
|
|
default:
|
|
return time.Hour
|
|
}
|
|
}
|
|
|
|
// ---- API: GetTraces GET /api/v1/traces ----
|
|
|
|
func (h *TraceProxyHandler) GetTraces(c *gin.Context) {
|
|
service := c.Query("service")
|
|
status := c.DefaultQuery("status", "all")
|
|
timeRange := c.DefaultQuery("timeRange", "1h")
|
|
limit := 50
|
|
|
|
ds, err := h.ds.getDefaultTraceDS()
|
|
if err != nil || ds == nil {
|
|
// demo 数据
|
|
demoTraces := demoTraces(service, status)
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
|
"data": demoTraces,
|
|
"total": len(demoTraces),
|
|
"demo": true,
|
|
}})
|
|
return
|
|
}
|
|
|
|
dur := parseTraceRange(timeRange)
|
|
end := time.Now()
|
|
start := end.Add(-dur)
|
|
traces := h.queryTraces(ds, service, limit, start, end)
|
|
|
|
items := []traceListItem{}
|
|
for _, t := range traces {
|
|
if len(t.Spans) == 0 {
|
|
continue
|
|
}
|
|
rootSpan := t.Spans[0]
|
|
svcName := ""
|
|
if proc, ok := t.Processes[rootSpan.ProcessID]; ok {
|
|
svcName = proc.ServiceName
|
|
}
|
|
st := spanStatus(t.Spans)
|
|
// 按状态过滤
|
|
if status != "all" && st != status {
|
|
continue
|
|
}
|
|
items = append(items, traceListItem{
|
|
ID: t.TraceID,
|
|
TraceID: t.TraceID,
|
|
ServiceName: svcName,
|
|
OperationName: rootSpan.OperationName,
|
|
StartTime: formatTraceTime(rootSpan.StartTime),
|
|
Duration: math.Round(float64(rootSpan.Duration)/1000.0*100) / 100, // μs→ms
|
|
Status: st,
|
|
SpanCount: len(t.Spans),
|
|
Tags: tagsToMap(rootSpan.Tags),
|
|
})
|
|
}
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
|
"data": items,
|
|
"total": len(items),
|
|
}})
|
|
}
|
|
|
|
// queryTraces 查询 Jaeger traces
|
|
func (h *TraceProxyHandler) queryTraces(ds *model.TraceDataSource, service string, limit int, start, end time.Time) []jaegerTrace {
|
|
path := fmt.Sprintf("/api/traces?limit=%d&start=%d&end=%d", limit, start.UnixMicro(), end.UnixMicro())
|
|
if service != "" && service != "all" {
|
|
path += "&service=" + service
|
|
}
|
|
body, status, err := h.doGet(ds, path)
|
|
if err != nil || status != 200 {
|
|
return nil
|
|
}
|
|
var resp jaegerTraceResp
|
|
if json.Unmarshal(body, &resp) != nil {
|
|
return nil
|
|
}
|
|
return resp.Data
|
|
}
|
|
|
|
// ---- API: GetTraceDetail GET /api/v1/traces/:traceId ----
|
|
|
|
func (h *TraceProxyHandler) GetTraceDetail(c *gin.Context) {
|
|
traceID := c.Param("traceId")
|
|
|
|
ds, err := h.ds.getDefaultTraceDS()
|
|
if err != nil || ds == nil {
|
|
// demo
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{
|
|
"spans": demoSpans(traceID),
|
|
"demo": true,
|
|
}})
|
|
return
|
|
}
|
|
|
|
body, status, err := h.doGet(ds, "/api/traces/"+traceID)
|
|
if err != nil || status != 200 {
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{"spans": []spanItem{}}})
|
|
return
|
|
}
|
|
|
|
var resp jaegerTraceResp
|
|
if json.Unmarshal(body, &resp) != nil || len(resp.Data) == 0 {
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{"spans": []spanItem{}}})
|
|
return
|
|
}
|
|
|
|
t := resp.Data[0]
|
|
spans := []spanItem{}
|
|
for _, sp := range t.Spans {
|
|
svcName := ""
|
|
if proc, ok := t.Processes[sp.ProcessID]; ok {
|
|
svcName = proc.ServiceName
|
|
}
|
|
parentID := ""
|
|
for _, ref := range sp.References {
|
|
if ref.RefType == "CHILD_OF" {
|
|
parentID = ref.SpanID
|
|
break
|
|
}
|
|
}
|
|
st := "success"
|
|
for _, tag := range sp.Tags {
|
|
if tag.Key == "error" {
|
|
if b, ok := tag.Value.(bool); ok && b {
|
|
st = "error"
|
|
}
|
|
}
|
|
}
|
|
logs := []spanLogItem{}
|
|
for _, l := range sp.Logs {
|
|
fields := map[string]string{}
|
|
for _, f := range l.Fields {
|
|
fields[f.Key] = fmt.Sprintf("%v", f.Value)
|
|
}
|
|
logs = append(logs, spanLogItem{
|
|
Timestamp: formatTraceTime(l.Timestamp),
|
|
Fields: fields,
|
|
})
|
|
}
|
|
spans = append(spans, spanItem{
|
|
ID: sp.SpanID,
|
|
SpanID: sp.SpanID,
|
|
ParentSpanID: parentID,
|
|
ServiceName: svcName,
|
|
OperationName: sp.OperationName,
|
|
StartTime: formatTraceTime(sp.StartTime),
|
|
Duration: math.Round(float64(sp.Duration)/1000.0*100) / 100,
|
|
Status: st,
|
|
Kind: kindFromTags(sp.Tags),
|
|
Tags: tagsToMap(sp.Tags),
|
|
Logs: logs,
|
|
})
|
|
}
|
|
c.JSON(200, model.Response{Code: 200, Data: gin.H{"spans": spans}})
|
|
}
|
|
|
|
// ---- Demo 数据 ----
|
|
|
|
func demoServiceMetrics() []serviceMetric {
|
|
return []serviceMetric{
|
|
{ServiceName: "api-gateway", RequestCount: 15230, ErrorRate: 0.32, AvgDuration: 45.2, P95Duration: 120.5, P99Duration: 250.1},
|
|
{ServiceName: "user-service", RequestCount: 8421, ErrorRate: 0.15, AvgDuration: 28.7, P95Duration: 75.3, P99Duration: 150.8},
|
|
{ServiceName: "order-service", RequestCount: 6754, ErrorRate: 1.23, AvgDuration: 89.4, P95Duration: 200.6, P99Duration: 450.2},
|
|
{ServiceName: "payment-service", RequestCount: 3210, ErrorRate: 0.08, AvgDuration: 156.3, P95Duration: 350.4, P99Duration: 680.7},
|
|
{ServiceName: "inventory-service", RequestCount: 4567, ErrorRate: 0.45, AvgDuration: 32.1, P95Duration: 85.7, P99Duration: 170.3},
|
|
}
|
|
}
|
|
|
|
func demoTraces(serviceFilter, statusFilter string) []traceListItem {
|
|
now := time.Now()
|
|
all := []traceListItem{
|
|
{ID: "trace-001", TraceID: "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4", ServiceName: "api-gateway", OperationName: "GET /api/v1/orders", StartTime: now.Add(-2 * time.Minute).Format("2006-01-02 15:04:05"), Duration: 125.4, Status: "success", SpanCount: 8, Tags: map[string]string{"http.method": "GET", "http.status_code": "200"}},
|
|
{ID: "trace-002", TraceID: "b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5", ServiceName: "user-service", OperationName: "POST /api/v1/users/login", StartTime: now.Add(-5 * time.Minute).Format("2006-01-02 15:04:05"), Duration: 45.7, Status: "success", SpanCount: 4, Tags: map[string]string{"http.method": "POST", "http.status_code": "200"}},
|
|
{ID: "trace-003", TraceID: "c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6", ServiceName: "order-service", OperationName: "POST /api/v1/orders", StartTime: now.Add(-8 * time.Minute).Format("2006-01-02 15:04:05"), Duration: 380.2, Status: "error", SpanCount: 12, Tags: map[string]string{"http.method": "POST", "http.status_code": "500", "error": "true"}},
|
|
{ID: "trace-004", TraceID: "d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1", ServiceName: "payment-service", OperationName: "POST /api/v1/payments", StartTime: now.Add(-12 * time.Minute).Format("2006-01-02 15:04:05"), Duration: 220.8, Status: "success", SpanCount: 6, Tags: map[string]string{"http.method": "POST", "http.status_code": "200"}},
|
|
{ID: "trace-005", TraceID: "e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2", ServiceName: "inventory-service", OperationName: "GET /api/v1/inventory/:id", StartTime: now.Add(-15 * time.Minute).Format("2006-01-02 15:04:05"), Duration: 32.1, Status: "success", SpanCount: 3, Tags: map[string]string{"http.method": "GET", "http.status_code": "200"}},
|
|
{ID: "trace-006", TraceID: "f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3", ServiceName: "api-gateway", OperationName: "GET /api/v1/products", StartTime: now.Add(-20 * time.Minute).Format("2006-01-02 15:04:05"), Duration: 95.3, Status: "success", SpanCount: 5, Tags: map[string]string{"http.method": "GET", "http.status_code": "200"}},
|
|
}
|
|
result := []traceListItem{}
|
|
for _, t := range all {
|
|
if serviceFilter != "" && serviceFilter != "all" && t.ServiceName != serviceFilter {
|
|
continue
|
|
}
|
|
if statusFilter != "all" && t.Status != statusFilter {
|
|
continue
|
|
}
|
|
result = append(result, t)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func demoSpans(traceID string) []spanItem {
|
|
now := time.Now().Add(-2 * time.Minute)
|
|
return []spanItem{
|
|
{ID: "span-1", SpanID: "span-1", ServiceName: "api-gateway", OperationName: "GET /api/v1/orders", StartTime: now.Format("2006-01-02 15:04:05"), Duration: 125.4, Status: "success", Kind: "server", Tags: map[string]string{"http.method": "GET", "http.url": "/api/v1/orders"}, Logs: []spanLogItem{}},
|
|
{ID: "span-2", SpanID: "span-2", ParentSpanID: "span-1", ServiceName: "user-service", OperationName: "auth.check", StartTime: now.Add(2 * time.Millisecond).Format("2006-01-02 15:04:05"), Duration: 15.2, Status: "success", Kind: "client", Tags: map[string]string{"rpc.system": "grpc"}, Logs: []spanLogItem{}},
|
|
{ID: "span-3", SpanID: "span-3", ParentSpanID: "span-1", ServiceName: "order-service", OperationName: "order.list", StartTime: now.Add(20 * time.Millisecond).Format("2006-01-02 15:04:05"), Duration: 65.8, Status: "success", Kind: "client", Tags: map[string]string{"db.system": "mysql", "db.statement": "SELECT"}, Logs: []spanLogItem{}},
|
|
{ID: "span-4", SpanID: "span-4", ParentSpanID: "span-3", ServiceName: "inventory-service", OperationName: "inventory.check", StartTime: now.Add(30 * time.Millisecond).Format("2006-01-02 15:04:05"), Duration: 22.1, Status: "success", Kind: "client", Tags: map[string]string{"rpc.system": "grpc"}, Logs: []spanLogItem{}},
|
|
{ID: "span-5", SpanID: "span-5", ParentSpanID: "span-3", ServiceName: "order-service", OperationName: "db.query", StartTime: now.Add(50 * time.Millisecond).Format("2006-01-02 15:04:05"), Duration: 10.5, Status: "success", Kind: "internal", Tags: map[string]string{"db.system": "mysql"}, Logs: []spanLogItem{}},
|
|
}
|
|
}
|
|
|
|
// 确保 strings 包被使用（避免 unused import 在某些构建条件下）
|
|
var _ = strings.TrimSpace
|