diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index 3fd4c2e..f9a2ccf 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -15,6 +15,7 @@ import ( "github.com/sundynix/sundynix-dispatcher/internal/llm" dnats "github.com/sundynix/sundynix-dispatcher/internal/nats" "github.com/sundynix/sundynix-shared/contract" + "github.com/sundynix/sundynix-shared/health" "github.com/sundynix/sundynix-shared/otelx" "github.com/sundynix/sundynix-shared/prompts" "github.com/sundynix/sundynix-shared/secrets" @@ -103,6 +104,15 @@ func main() { defer func() { _ = unsub() }() } + // HTTP 健康探针:给 k8s/LB 直接探(此前只有 NATS ServeHealth,编排器够不着)。 + // readiness = NATS 连接可用(能收任务);liveness = 进程能应答。 + healthShutdown := health.Serve("dispatcher", envOr("DISPATCHER_HEALTH_ADDR", ":8091"), sub.IsConnected) + defer func() { + sctx, scancel := context.WithTimeout(context.Background(), 5*time.Second) + defer scancel() + healthShutdown(sctx) + }() + // 监听退出信号,优雅停止消费。 ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go index d0f81cd..1c6f586 100644 --- a/sundynix-dispatcher/internal/nats/subscriber.go +++ b/sundynix-dispatcher/internal/nats/subscriber.go @@ -18,6 +18,9 @@ type Subscriber struct { inner *sharedbus.Bus } +// IsConnected 报告 NATS 此刻是否可用(供 readiness 探针)。 +func (s *Subscriber) IsConnected() bool { return s.inner.IsConnected() } + // MustConnect 接入 NATS 并确保任务流存在(消费者声明在 Consume 时完成)。 func MustConnect(url string) *Subscriber { inner, err := sharedbus.Connect(url) diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index 9a638f3..a1b431c 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -333,11 +333,20 @@ func (h *Handler) Healthz(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"status": "ok"}) } -// Readyz: GET /readyz —— 就绪探针(readiness):核心依赖(DB/Redis)可用才 200,否则 503。 -// 供 k8s 等编排器在依赖未就绪时暂不导流。NATS 在启动时即连(连不上会 fatal),故不单列。 +// Readyz: GET /readyz —— 就绪探针(readiness):**实时** ping 依赖,硬依赖 DB 可达才 200,否则 503。 +// 供 k8s/LB 在依赖未就绪或运行中掉线时暂不导流。 +// 两个刻意的设计: +// - 用 Ping 实时探活,**不是** Enabled() 启动期降级标志——后者反映不了「启动时连过、运行中 +// PG 掉线」,会让 LB 继续往已不可用的实例导流。 +// - 只把 **DB 当硬依赖门**:Redis 掉线仍可服务(限流有进程内 fail-safe 兜底、SSE 回落 live NATS), +// 若 Redis 一 blip 就把全部实例踢出轮转反而制造整站故障。Redis 只上报、不 gate。 +// NATS 启动即连(连不上 fatal),不单列。 func (h *Handler) Readyz(c *gin.Context) { - deps := gin.H{"db": h.db.Enabled(), "redis": h.cache.Enabled()} - if h.db.Enabled() && h.cache.Enabled() { + ctx, cancel := context.WithTimeout(c.Request.Context(), 2*time.Second) + defer cancel() + dbOK := h.db.Ping(ctx) + deps := gin.H{"db": dbOK, "redis": h.cache.Ping(ctx)} + if dbOK { c.JSON(http.StatusOK, gin.H{"status": "ready", "deps": deps}) return } diff --git a/sundynix-mcp-go/cmd/server/main.go b/sundynix-mcp-go/cmd/server/main.go index 32e1b8f..670ec75 100644 --- a/sundynix-mcp-go/cmd/server/main.go +++ b/sundynix-mcp-go/cmd/server/main.go @@ -7,10 +7,12 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/sundynix/sundynix-shared/blob" sharedbus "github.com/sundynix/sundynix-shared/bus" "github.com/sundynix/sundynix-shared/contract" + "github.com/sundynix/sundynix-shared/health" "github.com/sundynix/sundynix-mcp-go/internal/history" "github.com/sundynix/sundynix-mcp-go/internal/mcp" @@ -75,6 +77,15 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() + // HTTP 健康探针:给 k8s/LB 直接探(此前只有 NATS 应答,编排器够不着)。 + // readiness = NATS 连接可用(能收工具调用);liveness = 进程能应答。 + healthShutdown := health.Serve("mcp-go", envOr("MCP_GO_HEALTH_ADDR", ":8092"), b.IsConnected) + defer func() { + sctx, scancel := context.WithTimeout(context.Background(), 5*time.Second) + defer scancel() + healthShutdown(sctx) + }() + // RAG 核心链:embedding + Milvus(向量) + Bleve(全文) + Neo4j(图谱) + 可选 rerank ragEngine := rag.Open(ctx, rag.Config{ MilvusAddr: milvusAddr, diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index fc8fc4d..b266dd0 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -100,6 +100,9 @@ func waitConnected(nc *nats.Conn, d time.Duration) bool { return nc.Status() == nats.CONNECTED } +// IsConnected 报告 NATS 连接此刻是否真的可用(供 readiness 探针;非仅启动时连过)。 +func (b *Bus) IsConnected() bool { return b.nc != nil && b.nc.IsConnected() } + // Close 关闭底层连接。 func (b *Bus) Close() { if b.nc != nil { diff --git a/sundynix-shared/health/health.go b/sundynix-shared/health/health.go new file mode 100644 index 0000000..148f374 --- /dev/null +++ b/sundynix-shared/health/health.go @@ -0,0 +1,54 @@ +// Package health 提供轻量 HTTP 健康探针,给无 HTTP 端口的后端服务(dispatcher / mcp-go) +// 补上 k8s/LB 能直接探的 /healthz(liveness) 与 /readyz(readiness)。 +// +// 此前这两个服务只有 NATS ServeHealth 应答器,编排器无法对它们做 HTTP 探测、只能靠 gateway +// 经 NATS 代探。这里给它们各起一个极小的 HTTP 服务。 +package health + +import ( + "context" + "log" + "net/http" + "time" +) + +// Serve 起一个健康探针 HTTP 服务: +// - /healthz 恒 200(liveness:进程能应答即存活); +// - /readyz 由 ready() 决定 200/503(readiness:依赖就绪才导流)。 +// +// addr 为空则不启动、返回 no-op(本地无端口需求时)。返回 shutdown 供优雅停机调用。 +func Serve(service, addr string, ready func() bool) func(context.Context) { + if addr == "" { + return func(context.Context) {} + } + srv := &http.Server{Addr: addr, Handler: Handler(service, ready), ReadHeaderTimeout: 5 * time.Second} + go func() { + log.Printf("[%s] health probe on %s (/healthz /readyz)", service, addr) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("[%s] health probe listen: %v", service, err) + } + }() + return func(ctx context.Context) { _ = srv.Shutdown(ctx) } +} + +// Handler 返回探针路由(/healthz 恒 200,/readyz 由 ready 决定),便于单测与自定义挂载。 +func Handler(service string, ready func() bool) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, `{"status":"ok","service":"`+service+`"}`) + }) + mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) { + if ready == nil || ready() { + writeJSON(w, http.StatusOK, `{"status":"ready"}`) + return + } + writeJSON(w, http.StatusServiceUnavailable, `{"status":"not_ready"}`) + }) + return mux +} + +func writeJSON(w http.ResponseWriter, code int, body string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _, _ = w.Write([]byte(body)) +} diff --git a/sundynix-shared/health/health_test.go b/sundynix-shared/health/health_test.go new file mode 100644 index 0000000..2216e43 --- /dev/null +++ b/sundynix-shared/health/health_test.go @@ -0,0 +1,30 @@ +package health + +import ( + "net/http" + "net/http/httptest" + "testing" +) + +// /healthz 恒 200;/readyz 随 ready() 翻 200/503。 +func TestHandler(t *testing.T) { + ready := false + h := Handler("test", func() bool { return ready }) + + do := func(path string) int { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil)) + return w.Code + } + + if c := do("/healthz"); c != 200 { + t.Fatalf("/healthz 应恒 200,得 %d", c) + } + if c := do("/readyz"); c != 503 { + t.Fatalf("未就绪时 /readyz 应 503,得 %d", c) + } + ready = true + if c := do("/readyz"); c != 200 { + t.Fatalf("就绪后 /readyz 应 200,得 %d", c) + } +}