From 9b153871eb9ab151f6ca8337d248ee84797e43a6 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 21 Jul 2026 16:15:22 +0800 Subject: [PATCH] =?UTF-8?q?feat(monitor):=20NATS=20=E9=9B=86=E7=BE=A4=20Ra?= =?UTF-8?q?ft=20=E5=89=AF=E6=9C=AC=E5=81=A5=E5=BA=B7=20+=20=E5=9F=BA?= =?UTF-8?q?=E5=BB=BA=20ping=20=E5=BB=B6=E8=BF=9F(=E7=9B=91=E6=B5=8B?= =?UTF-8?q?=E7=BB=84=E5=AE=8C=E5=96=84)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前 /status 把 NATS 当一盏二元灯(连不上就 fatal 故恒真)，看不出 3 节点集群里 哪个节点掉了、JetStream 持久流的 Raft 副本是否还齐(计费/状态/评测流不丢的关键)。 DB/Redis/MinIO 也只二元 ping、无延迟。 - bus.ClusterStatus：连的节点名 + 集群发现节点数(nc.Servers) + RTT(nc.RTT) + 6 条关键 持久流(tasks/status/usage/eval/ingest/approvals)的 Raft 副本健康(leader + healthy/total， 单节点部署记 1/1；某节点掉队 → healthy --- sundynix-admin/src/api.ts | 16 ++++ sundynix-admin/src/pages/StatusPage.tsx | 65 +++++++++++++++++ .../internal/handler/status_handler.go | 73 +++++++++++++++---- sundynix-gateway/internal/nats/publisher.go | 5 ++ sundynix-shared/bus/bus.go | 72 ++++++++++++++++++ 5 files changed, 217 insertions(+), 14 deletions(-) diff --git a/sundynix-admin/src/api.ts b/sundynix-admin/src/api.ts index 03618aa..9e6295e 100644 --- a/sundynix-admin/src/api.ts +++ b/sundynix-admin/src/api.ts @@ -670,11 +670,27 @@ export interface ToolGroup { up: boolean; tools: ToolInfo[] | null; } +export interface NatsStreamHealth { + name: string; + leader: string; + replicas_healthy: number; + replicas_total: number; + messages: number; +} +export interface NatsClusterStatus { + connected: boolean; + connected_to: string; + known_servers: number; + rtt_ms: number; + streams: NatsStreamHealth[] | null; + degraded: number; +} export interface SystemStatus { checked_at: string; infra: StatusItem[]; services: StatusItem[]; tools: ToolGroup[]; + nats?: NatsClusterStatus; } export async function getStatus(): Promise { diff --git a/sundynix-admin/src/pages/StatusPage.tsx b/sundynix-admin/src/pages/StatusPage.tsx index bfacc44..e540bce 100644 --- a/sundynix-admin/src/pages/StatusPage.tsx +++ b/sundynix-admin/src/pages/StatusPage.tsx @@ -330,6 +330,9 @@ export function StatusPage() { )}
+ {item.up && item.latency_ms != null && item.latency_ms > 0 && ( + {item.latency_ms}ms + )} :{meta?.port || "—"} @@ -344,6 +347,68 @@ export function StatusPage() {
+ {/* NATS 集群 · JetStream Raft 副本健康 */} + {data.nats && ( +
+
+
+

+ + NATS 消息骨干 · 集群 +

+

JetStream Raft 副本健康——计费/状态/评测等持久流不丢的保证

+
+
+ {[ + { k: "节点", v: String(data.nats.known_servers) }, + { k: "RTT", v: `${data.nats.rtt_ms}ms` }, + { k: "连接", v: data.nats.connected_to || "—" }, + ].map((m) => ( +
+
{m.k}
+
{m.v}
+
+ ))} +
+
+
+ + + + + + + + + + + {(data.nats.streams ?? []).map((s) => { + const ok = s.replicas_healthy >= s.replicas_total; + return ( + + + + + + + ); + })} + {(data.nats.streams ?? []).length === 0 && ( + + + + )} + +
持久流Leader副本健康消息数
{s.name}{s.leader || "—"} + + {s.replicas_healthy}/{s.replicas_total} + + {!ok && 降级} + {s.messages.toLocaleString()}
暂无流信息
+
+
+ )} + {/* MCP 工具注册箱 */}
diff --git a/sundynix-gateway/internal/handler/status_handler.go b/sundynix-gateway/internal/handler/status_handler.go index 1fe624a..36573ed 100644 --- a/sundynix-gateway/internal/handler/status_handler.go +++ b/sundynix-gateway/internal/handler/status_handler.go @@ -10,6 +10,7 @@ import ( "github.com/gin-gonic/gin" + sharedbus "github.com/sundynix/sundynix-shared/bus" "github.com/sundynix/sundynix-shared/contract" ) @@ -35,12 +36,13 @@ type toolGroup struct { Tools []toolInfo `json:"tools"` } -// systemStatus 是「服务状态」面板的聚合视图：基建 / 应用服务 / MCP 工具注册。 +// systemStatus 是「服务状态」面板的聚合视图：基建 / 应用服务 / MCP 工具注册 / NATS 集群。 type systemStatus struct { - CheckedAt string `json:"checked_at"` - Infra []statusItem `json:"infra"` - Services []statusItem `json:"services"` - Tools []toolGroup `json:"tools"` + 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 是各探针的单次超时（无响应即判为下线）。 @@ -68,9 +70,11 @@ func (h *Handler) AdminStatus(c *gin.Context) { dispLatency int // dispatcher 探针耗时 pgUp, redisUp, minioUp bool // 基建活性探针（实时 ping，非仅启动标志） + pgMs, redisMs, minioMs int // 各基建 ping 往返耗时（ms） + natsClu sharedbus.NATSClusterStatus ) - wg.Add(5) + wg.Add(6) // 1) mcp-go health → milvus / neo4j 基建灯 go func() { @@ -127,15 +131,25 @@ func (h *Handler) AdminStatus(c *gin.Context) { }) }() - // 5) 基建活性探针：PG / Redis / MinIO 实时 ping（非仅启动标志，能反映中途掉线）。 + // 5) 基建活性探针：PG / Redis / MinIO 实时 ping（非仅启动标志，能反映中途掉线）+ 往返耗时。 go func() { defer wg.Done() safeCall("status-probe-infra", func() { ctx, cancel := context.WithTimeout(parent, probeTimeout) defer cancel() - pgUp = h.db.Ping(ctx) - redisUp = h.cache.Ping(ctx) - minioUp = h.blob != nil && h.blob.Ping(ctx) + 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) }) }() @@ -144,16 +158,18 @@ func (h *Handler) AdminStatus(c *gin.Context) { c.JSON(http.StatusOK, systemStatus{ CheckedAt: time.Now().Format(time.RFC3339), Infra: []statusItem{ - {Name: "postgres", Up: pgUp}, - {Name: "redis", Up: redisUp}, - {Name: "nats", Up: true}, // 网关连不上 NATS 即 fatal，能应答即在线 + {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}, // 对象存储（报告/KB 正文/blob，126） + {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}, @@ -183,6 +199,35 @@ func (h *Handler) probeTools(parent context.Context, subject string) (up bool, t 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 != "" { diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go index 1223461..8a070f1 100644 --- a/sundynix-gateway/internal/nats/publisher.go +++ b/sundynix-gateway/internal/nats/publisher.go @@ -40,6 +40,11 @@ func MustConnect(url string) *Bus { return &Bus{inner: inner} } +// ClusterStatus 透传共享 bus 的 NATS 集群体检（供「服务状态」监测面板）。 +func (b *Bus) ClusterStatus(ctx context.Context) sharedbus.NATSClusterStatus { + return b.inner.ClusterStatus(ctx) +} + // PublishTask 把组装后的 Task 发布到 sundynix.tasks.。 func (b *Bus) PublishTask(ctx context.Context, t *contract.Task) error { seq, err := b.inner.PublishTask(ctx, t) diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index b266dd0..9e71ce8 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -103,6 +103,78 @@ func waitConnected(nc *nats.Conn, d time.Duration) bool { // IsConnected 报告 NATS 连接此刻是否真的可用（供 readiness 探针；非仅启动时连过）。 func (b *Bus) IsConnected() bool { return b.nc != nil && b.nc.IsConnected() } +// StreamHealth 是一条 JetStream 流的 Raft 副本健康。单节点部署 Cluster 为空 → 记 1/1。 +type StreamHealth struct { + Name string `json:"name"` + Leader string `json:"leader"` + ReplicasHealthy int `json:"replicas_healthy"` + ReplicasTotal int `json:"replicas_total"` + Messages uint64 `json:"messages"` +} + +// NATSClusterStatus 是 NATS 骨干网的实时集群视图（供监测面板：不再是一盏二元灯）。 +type NATSClusterStatus struct { + Connected bool `json:"connected"` + ConnectedTo string `json:"connected_to"` // 当前连的节点名 + KnownServers int `json:"known_servers"` // 集群发现到的节点数（3 节点集群应为 3） + RTTMillis int `json:"rtt_ms"` // 到当前节点的往返耗时 + Streams []StreamHealth `json:"streams"` // 关键持久流的副本健康 + Degraded int `json:"degraded"` // 副本未满（有节点掉队）的流数 +} + +// clusterStreams 是要体检的关键持久流（计费/状态/评测/任务/入库/审批都不能丢）。 +var clusterStreams = []string{ + contract.StreamTasks, contract.StreamStatus, contract.StreamUsage, + contract.StreamEval, contract.StreamIngest, contract.StreamApprovals, +} + +// ClusterStatus 探 NATS 集群实时状态：连的节点、已知节点数、RTT，以及各关键流的 Raft 副本健康。 +// 3 节点集群里某节点掉线 → 对应流 ReplicasHealthy < Total（仍有 quorum 可服务，但已降级须告警）。 +func (b *Bus) ClusterStatus(ctx context.Context) NATSClusterStatus { + st := NATSClusterStatus{} + if b.nc == nil { + return st + } + st.Connected = b.nc.IsConnected() + st.ConnectedTo = b.nc.ConnectedServerName() + st.KnownServers = len(b.nc.Servers()) + if rtt, err := b.nc.RTT(); err == nil { + st.RTTMillis = int(rtt.Milliseconds()) + } + for _, name := range clusterStreams { + s, err := b.js.Stream(ctx, name) + if err != nil { + continue + } + info, err := s.Info(ctx) + if err != nil { + continue + } + sh := StreamHealth{Name: name, Messages: info.State.Msgs} + if info.Cluster != nil { + sh.Leader = info.Cluster.Leader + total := 1 + len(info.Cluster.Replicas) // leader + 副本 + healthy := 0 + if sh.Leader != "" { + healthy++ // leader 在即算健康 1 个 + } + for _, p := range info.Cluster.Replicas { + if p.Current && !p.Offline { + healthy++ + } + } + sh.ReplicasHealthy, sh.ReplicasTotal = healthy, total + } else { + sh.ReplicasHealthy, sh.ReplicasTotal, sh.Leader = 1, 1, st.ConnectedTo // 单节点 + } + if sh.ReplicasHealthy < sh.ReplicasTotal { + st.Degraded++ + } + st.Streams = append(st.Streams, sh) + } + return st +} + // Close 关闭底层连接。 func (b *Bus) Close() { if b.nc != nil {