sundynix/sundynix-agentix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(monitor): NATS 集群 Raft 副本健康 + 基建 ping 延迟(监测组完善)

Browse Source 
此前 /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<total 记降级)。
- /status 新增 nats 集群对象 + NATS 灯改为'连接且无副本降级才绿'、detail 显示'N 节点·连 X·
  流副本齐全/降级'、latency=RTT；PG/Redis/MinIO 加 ping 往返耗时。
- admin StatusPage：新增 NATS 集群面板(节点/RTT/连接 + 各流 leader/副本健康/消息数表)，
  基建行显示延迟。

sundynix-shared/gateway build+vet+test 绿；admin tsc 绿。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-07-21 16:15:22 +08:00
parent a42f30239f
commit 9b153871eb
5 changed files with 217 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sundynix-admin/src/api.ts
+16
View File
@@ -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<SystemStatus> {
sundynix-admin/src/pages/StatusPage.tsx
+65
View File
@@ -330,6 +330,9 @@ export function StatusPage() {
)}
</div>
<div className="flex items-center gap-3">
{item.up && item.latency_ms != null && item.latency_ms > 0 && (
<span className="font-mono text-[10px] text-slate-400 tabular-nums">{item.latency_ms}ms</span>
)}
<code className="text-[10px] text-slate-400 font-mono bg-slate-100 px-1.5 py-0.5 rounded">
:{meta?.port || "—"}
</code>
@@ -344,6 +347,68 @@ export function StatusPage() {
</section>
</div>
{/* NATS 集群 · JetStream Raft 副本健康 */}
{data.nats && (
<section className="rounded-2xl border border-gray-200 bg-white p-5">
<div className="border-b border-slate-100 pb-3.5 mb-4 flex flex-wrap items-center justify-between gap-3">
<div>
<h3 className="text-sm font-bold text-slate-800 flex items-center gap-1.5">
<span className="h-2 w-2 rounded-full bg-indigo-500"></span>
NATS 消息骨干 · 集群
</h3>
<p className="text-[10px] text-slate-400 mt-0.5">JetStream Raft 副本健康——计费/状态/评测等持久流不丢的保证</p>
</div>
<div className="flex items-center gap-5 text-xs">
{[
{ k: "节点", v: String(data.nats.known_servers) },
{ k: "RTT", v: `${data.nats.rtt_ms}ms` },
{ k: "连接", v: data.nats.connected_to || "—" },
].map((m) => (
<div key={m.k} className="text-right">
<div className="text-[9px] uppercase text-slate-400 tracking-wider font-semibold">{m.k}</div>
<div className="font-bold text-slate-700 font-mono truncate max-w-[140px]" title={m.v}>{m.v}</div>
</div>
))}
</div>
</div>
<div className="overflow-x-auto">
<table className="w-full text-xs">
<thead>
<tr className="text-left text-[10px] uppercase tracking-wide text-slate-400">
<th className="pb-2 font-medium">持久流</th>
<th className="pb-2 font-medium">Leader</th>
<th className="pb-2 font-medium text-right">副本健康</th>
<th className="pb-2 font-medium text-right">消息数</th>
</tr>
</thead>
<tbody>
{(data.nats.streams ?? []).map((s) => {
const ok = s.replicas_healthy >= s.replicas_total;
return (
<tr key={s.name} className="border-t border-slate-50">
<td className="py-2 font-mono text-slate-700">{s.name}</td>
<td className="py-2 font-mono text-[10px] text-slate-500 truncate max-w-[160px]" title={s.leader}>{s.leader || "—"}</td>
<td className="py-2 text-right">
<span className={`font-bold font-mono ${ok ? "text-emerald-600" : "text-amber-600"}`}>
{s.replicas_healthy}/{s.replicas_total}
</span>
{!ok && <span className="ml-1 text-[9px] text-amber-600">降级</span>}
</td>
<td className="py-2 text-right font-mono text-slate-500 tabular-nums">{s.messages.toLocaleString()}</td>
</tr>
);
})}
{(data.nats.streams ?? []).length === 0 && (
<tr>
<td colSpan={4} className="py-4 text-center text-[10px] text-slate-400">暂无流信息</td>
</tr>
)}
</tbody>
</table>
</div>
</section>
)}
{/* MCP 工具注册箱 */}
<section className="rounded-2xl border border-gray-200 bg-white p-6 space-y-4">
<div className="border-b border-slate-100 pb-3 flex flex-wrap items-center justify-between gap-3">
sundynix-gateway/internal/handler/status_handler.go
+59 -14
View File
@@ -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 != "" {
sundynix-gateway/internal/nats/publisher.go
+5
View File
@@ -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.<id>。
func (b *Bus) PublishTask(ctx context.Context, t *contract.Task) error {
seq, err := b.inner.PublishTask(ctx, t)
sundynix-shared/bus/bus.go
+72
View File
@@ -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 {