package handler import ( "context" "encoding/json" "fmt" "net/http" "sync" "time" "github.com/gin-gonic/gin" sharedbus "github.com/sundynix/sundynix-shared/bus" "github.com/sundynix/sundynix-shared/contract" ) // statusItem 是一项依赖/服务的存活状态(基建灯、服务灯共用)。 type statusItem struct { Name string `json:"name"` Up bool `json:"up"` Detail string `json:"detail,omitempty"` Latency int `json:"latency_ms,omitempty"` // NATS 探针往返耗时(毫秒),本地检查项为 0 } // toolInfo 是一个注册工具的元信息(透传各 MCP 服务 list_tools 的上报)。 type toolInfo struct { Name string `json:"name"` CN string `json:"cn"` Desc string `json:"desc"` } // toolGroup 是一台 MCP 服务的工具注册情况。 type toolGroup struct { Server string `json:"server"` Up bool `json:"up"` Tools []toolInfo `json:"tools"` } // systemStatus 是「服务状态」面板的聚合视图:基建 / 应用服务 / MCP 工具注册 / NATS 集群。 type systemStatus struct { CheckedAt string `json:"checked_at"` Infra []statusItem `json:"infra"` Services []statusItem `json:"services"` Tools []toolGroup `json:"tools"` Nats *sharedbus.NATSClusterStatus `json:"nats,omitempty"` // NATS 集群详情(节点/RTT/流副本 Raft 健康) } // probeTimeout 是各探针的单次超时(无响应即判为下线)。 const probeTimeout = 2 * time.Second // AdminStatus: GET /api/v1/admin/status —— 聚合基建、应用服务与 MCP 工具注册的实时状态, // 供管理端「服务状态」一眼看出哪个服务没起、基建是否就绪、工具是否注册。 func (h *Handler) AdminStatus(c *gin.Context) { parent := c.Request.Context() // 各探针互不依赖,并发执行,整体只等最慢的一个。 var ( wg sync.WaitGroup milvus, neo4j bool // mcp-go health ftDisk bool // 全文索引是否落盘持久(false=退内存兜底,重启清零) goUp bool // mcp-go 在线 goTools []toolInfo // mcp-go 注册工具 goLatency int // mcp-go 探针耗时 pyUp bool // mcp-py 在线 pyTools []toolInfo // mcp-py 注册工具 pyLatency int // mcp-py 探针耗时 dispUp bool // dispatcher 在线 dispDetail string // dispatcher 详情(模型/运行时长) dispLatency int // dispatcher 探针耗时 pgUp, redisUp, minioUp bool // 基建活性探针(实时 ping,非仅启动标志) pgMs, redisMs, minioMs int // 各基建 ping 往返耗时(ms) natsClu sharedbus.NATSClusterStatus ) wg.Add(6) // 1) mcp-go health → milvus / neo4j 基建灯 go func() { defer wg.Done() safeCall("status-probe-mcpgo-health", func() { ctx, cancel := context.WithTimeout(parent, probeTimeout) defer cancel() if res, err := h.bus.CallTool(ctx, contract.ToolSubjectGo("health"), &contract.ToolCall{Tool: "health"}); err == nil && res != nil && res.OK { var sub map[string]bool if json.Unmarshal([]byte(res.Content), &sub) == nil { milvus, neo4j = sub["milvus"], sub["neo4j"] ftDisk = sub["fulltext_disk"] } } }) }() // 2) mcp-go list_tools → 在线判定 + 工具清单 go func() { defer wg.Done() safeCall("status-probe-mcpgo-tools", func() { goUp, goTools, goLatency = h.probeTools(parent, contract.ToolSubjectGo("list_tools")) }) }() // 3) mcp-py list_tools → 在线判定 + 工具清单 go func() { defer wg.Done() safeCall("status-probe-mcppy-tools", func() { pyUp, pyTools, pyLatency = h.probeTools(parent, contract.ToolSubjectPy("list_tools")) }) }() // 4) dispatcher 心跳 → 在线判定 + 模型/运行时长 go func() { defer wg.Done() safeCall("status-probe-dispatcher", func() { ctx, cancel := context.WithTimeout(parent, probeTimeout) defer cancel() start := time.Now() if data, err := h.bus.Ping(ctx, contract.SubjectHealthDispatcher); err == nil { dispUp = true dispLatency = int(time.Since(start).Milliseconds()) var st struct { Model string `json:"model"` Ready bool `json:"ready"` UptimeS int `json:"uptime_s"` } if json.Unmarshal(data, &st) == nil { dispDetail = dispatcherDetail(st.Model, st.Ready, st.UptimeS) } } }) }() // 5) 基建活性探针:PG / Redis / MinIO 实时 ping(非仅启动标志,能反映中途掉线)+ 往返耗时。 go func() { defer wg.Done() safeCall("status-probe-infra", func() { ctx, cancel := context.WithTimeout(parent, probeTimeout) defer cancel() pgUp, pgMs = pingLatency(func() bool { return h.db.Ping(ctx) }) redisUp, redisMs = pingLatency(func() bool { return h.cache.Ping(ctx) }) minioUp, minioMs = pingLatency(func() bool { return h.blob != nil && h.blob.Ping(ctx) }) }) }() // 6) NATS 集群体检:节点数 / RTT / 各关键流 Raft 副本健康(不再是一盏二元灯)。 go func() { defer wg.Done() safeCall("status-probe-nats", func() { ctx, cancel := context.WithTimeout(parent, probeTimeout) defer cancel() natsClu = h.bus.ClusterStatus(ctx) }) }() wg.Wait() c.JSON(http.StatusOK, systemStatus{ CheckedAt: time.Now().Format(time.RFC3339), Infra: []statusItem{ {Name: "postgres", Up: pgUp, Latency: pgMs}, {Name: "redis", Up: redisUp, Latency: redisMs}, // NATS:不再是二元灯——上报连的节点、集群节点数、RTT、有几条流副本降级。 {Name: "nats", Up: natsClu.Connected && natsClu.Degraded == 0, Detail: natsDetail(natsClu), Latency: natsClu.RTTMillis}, {Name: "milvus", Up: milvus}, {Name: "neo4j", Up: neo4j}, {Name: "minio", Up: minioUp, Latency: minioMs}, // 对象存储(报告/KB 正文/blob,126) // 全文索引:mcp-go 本地 bleve,是唯一不在 128 集中存储上的检索路, // 也是唯一会"静默降级"的一路(退内存后重启清零,检索只是变差不报错)。 {Name: "全文索引", Up: goUp && ftDisk, Detail: fulltextDetail(goUp, ftDisk)}, }, Nats: &natsClu, Services: []statusItem{ {Name: "gateway", Up: true, Detail: "在线"}, {Name: "dispatcher", Up: dispUp, Detail: serviceDetail(dispUp, dispDetail), Latency: dispLatency}, {Name: "mcp-go", Up: goUp, Detail: toolsDetail(goUp, len(goTools)), Latency: goLatency}, {Name: "mcp-py", Up: pyUp, Detail: toolsDetail(pyUp, len(pyTools)), Latency: pyLatency}, }, Tools: []toolGroup{ {Server: "mcp-go", Up: goUp, Tools: goTools}, {Server: "mcp-py", Up: pyUp, Tools: pyTools}, }, }) } // probeTools 调一台 MCP 服务的 list_tools:能应答即在线,并解析其工具清单(含中文名/作用)+ 往返耗时。 func (h *Handler) probeTools(parent context.Context, subject string) (up bool, tools []toolInfo, latency int) { ctx, cancel := context.WithTimeout(parent, probeTimeout) defer cancel() start := time.Now() res, err := h.bus.CallTool(ctx, subject, &contract.ToolCall{Tool: "list_tools"}) if err != nil || res == nil || !res.OK { return false, nil, 0 } var payload struct { Tools []toolInfo `json:"tools"` } _ = json.Unmarshal([]byte(res.Content), &payload) return true, payload.Tools, int(time.Since(start).Milliseconds()) } // pingLatency 跑一次 ping 并计时,返回 (是否可达, 往返毫秒)。不可达则耗时记 0。 func pingLatency(ping func() bool) (bool, int) { start := time.Now() ok := ping() if !ok { return false, 0 } return true, int(time.Since(start).Milliseconds()) } // natsDetail 把 NATS 集群体检拼成一行人读摘要:节点数 · 连的节点 · 流副本健康。 func natsDetail(s sharedbus.NATSClusterStatus) string { if !s.Connected { return "未连接" } d := fmt.Sprintf("%d 节点", s.KnownServers) if s.ConnectedTo != "" { d += " · 连 " + s.ConnectedTo } if n := len(s.Streams); n > 0 { if s.Degraded > 0 { d += fmt.Sprintf(" · %d/%d 流副本降级(有节点掉队)", s.Degraded, n) } else { d += fmt.Sprintf(" · %d 流副本齐全", n) } } return d } func dispatcherDetail(model string, ready bool, uptimeS int) string { d := "运行 " + humanDuration(uptimeS) if model != "" { d = "模型 " + model + " · " + d } if !ready { d += "(模型未配置,降级桩)" } return d } func serviceDetail(up bool, detail string) string { if !up { return "无响应(未启动?)" } if detail == "" { return "在线" } return detail } func toolsDetail(up bool, n int) string { if !up { return "无响应(未启动?)" } return fmt.Sprintf("%d 个工具", n) } // humanDuration 把秒数转人读(< 1h 显示分钟,否则小时+分钟)。 func humanDuration(s int) string { if s < 60 { return fmt.Sprintf("%ds", s) } m := s / 60 if m < 60 { return fmt.Sprintf("%dm", m) } return fmt.Sprintf("%dh%dm", m/60, m%60) } // fulltextDetail 说明全文(bleve)索引的持久化状态。退内存兜底时必须讲清后果—— // 否则一盏灰灯没人知道意味着"重启就没了"。 func fulltextDetail(goUp, disk bool) string { switch { case !goUp: return "mcp-go 离线,无法判定" case disk: return "落盘持久" default: return "内存兜底 · 重启即清零(检查 BLEVE_PATH 与挂载卷权限)" } }