From 241c27358bc8f09f7bad832bac63e0016a53fd37 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 09:47:55 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat=E2=9C=A8:=20answer=20readiness=20separ?= =?UTF-8?q?ately=20from=20liveness,=20and=20fail=20it=20while=20draining?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /health returned 200 without asking anything. Whatever it was meant to say, an orchestrator reading it learned only that a process was accepting connections. The two questions are not the same one, and the answers differ: - /health stays a bare 200. It answers "should I restart you", and a process whose database is unreachable does not want restarting - that turns one outage into a crash loop and discards the connection pool, the cache and every request in flight on the way. - /ready is new. It answers "should I send you requests", fails while a dependency is unreachable, and fails from the moment shutdown begins. That last part is what the life-cycle phases bought. BeginDraining sits next to BeginShutdown, before the server stops accepting, so a load balancer is told to stop sending while this instance can still finish what it holds. Reversed - and that is where it was - the connections are cut first and the probe reports it afterwards. The queue is deliberately not checked. Nothing on AdapterQueue answers "are you reachable" without publishing something, the memory backend cannot fail, and a queue that is down degrades logging rather than stopping requests: a reason to alert, not a reason to leave the pool. The cache probe writes and reads back rather than only reading. A cache that answers "miss" for every key - a client pointed at the wrong server - is indistinguishable from a healthy one on a read alone. Every check runs behind a recover, and that is not defensive habit. The test for "nothing configured" found the reason: GetCacheAdapter builds a wrapper around whatever is configured and returns it even when nothing is, so the value is not nil, the cache inside it is, and Set dereferences it. A nil check cannot see that, and GetQueueAdapter behaves the same way. Whatever the cause, a probe is the last thing that should be able to take the process down - the caller is asking whether this instance is well, and killing it is the wrong reply. The counter-proof compiles and fails: without the recover, the unconfigured case panics rather than reporting two failed checks. --- app/other/router/monitor.go | 46 ++++++++++- cmd/api/server.go | 7 ++ common/health/health.go | 133 ++++++++++++++++++++++++++++++++ common/health/health_test.go | 141 ++++++++++++++++++++++++++++++++++ common/middleware/settings.go | 1 + 5 files changed, 325 insertions(+), 3 deletions(-) create mode 100644 common/health/health.go create mode 100644 common/health/health_test.go diff --git a/app/other/router/monitor.go b/app/other/router/monitor.go index 9c088992..ed48ea7d 100644 --- a/app/other/router/monitor.go +++ b/app/other/router/monitor.go @@ -1,23 +1,63 @@ package router import ( + "context" "net/http" + "time" "github.com/gin-gonic/gin" "github.com/go-admin-team/go-admin-core/v2/tools/transfer" "github.com/prometheus/client_golang/prometheus/promhttp" + + "go-admin/common/health" ) func init() { routerNoCheckRole = append(routerNoCheckRole, registerMonitorRouter) } -// 需认证的路由代码 +// readyTimeout bounds the whole probe. It has to stay under whatever period +// the orchestrator polls on, or a slow dependency turns a readiness check into +// a queue of readiness checks. +const readyTimeout = 2 * time.Second + +// 无需认证的路由代码 func registerMonitorRouter(v1 *gin.RouterGroup) { v1.GET("/metrics", transfer.Handler(promhttp.Handler())) - //健康检查 + + // 健康检查（存活） + // + // Stays a bare 200 on purpose. This is the answer to "should I restart + // you", and a process whose database is unreachable does not want + // restarting - that turns one outage into a crash loop and throws away the + // connection pool, the cache and every in-flight request along the way. v1.GET("/health", func(c *gin.Context) { c.Status(http.StatusOK) }) -} \ No newline at end of file + // 就绪检查 + // + // The answer to "should I send you requests". It fails while a dependency + // is unreachable, and from the moment shutdown begins - which is before + // the server stops accepting, so a load balancer can take this instance + // out of the pool while it can still finish what it has. + v1.GET("/ready", func(c *gin.Context) { + if health.Draining() { + c.JSON(http.StatusServiceUnavailable, gin.H{ + "status": "draining", + "checks": []health.Check{}, + }) + return + } + ctx, cancel := context.WithTimeout(c.Request.Context(), readyTimeout) + defer cancel() + + checks := health.Ready(ctx) + status := http.StatusOK + if !health.Healthy(checks) { + status = http.StatusServiceUnavailable + } + c.JSON(status, gin.H{"status": http.StatusText(status), "checks": checks}) + }) + +} diff --git a/cmd/api/server.go b/cmd/api/server.go index 2237ac76..334a6c27 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -30,6 +30,7 @@ import ( "go-admin/app/jobs" "go-admin/common/database" "go-admin/common/global" + "go-admin/common/health" common "go-admin/common/middleware" "go-admin/common/middleware/handler" "go-admin/common/storage" @@ -217,6 +218,12 @@ func run() error { // and the queue adapter, and re-registering consumers - on top of cleanup // that has already run. sdk.Runtime.BeginShutdown() + // Readiness fails from here, which is before the server stops accepting. + // The order is the whole point: a load balancer that is told "not ready" + // while this instance can still finish what it has in flight takes it out + // of the pool without dropping anything. Reversed, the connections are cut + // first and the health check reports it afterwards. + health.BeginDraining() log.Info("Shutdown Server ... ") if err := shutdownServer(srv, shutdownTimeout); err != nil { diff --git a/common/health/health.go b/common/health/health.go new file mode 100644 index 00000000..79c8ec01 --- /dev/null +++ b/common/health/health.go @@ -0,0 +1,133 @@ +// Package health answers whether this process should be sent traffic. +// +// The two questions an orchestrator asks are not the same one, and go-admin +// answers them at two endpoints: +// +// - /health is liveness: is the process there at all. It stays a bare 200, +// because the honest answer to "should I restart you" is almost always no. +// Restarting a process because its database is unreachable turns one +// outage into a crash loop that also loses the connection pool, the cache +// and every in-flight request. +// - /ready is readiness: should this instance receive requests now. It fails +// while the dependencies are unreachable, and - the part that only exists +// because of the life-cycle phases - it fails as soon as shutdown begins, +// before the server stops accepting, so a load balancer has a chance to +// take the instance out before connections are cut. +package health + +import ( + "context" + "errors" + "fmt" + "sync/atomic" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" +) + +// draining is set when the process starts shutting down. +// +// It is kept here rather than read back from core: BeginShutdown sets a flag on +// the Application, but nothing exports it, and one host wanting to know is not +// yet a reason to widen that interface. +var draining atomic.Bool + +// BeginDraining records that shutdown has started, so readiness fails from now +// on. It is called with BeginShutdown, before anything is taken apart. +func BeginDraining() { draining.Store(true) } + +// Draining reports whether shutdown has begun. +func Draining() bool { return draining.Load() } + +// Check is one dependency and what asking it produced. +type Check struct { + Name string `json:"name"` + OK bool `json:"ok"` + Err string `json:"error,omitempty"` +} + +// Ready asks every dependency this process cannot serve a request without. +// +// The queue is deliberately absent. Nothing on AdapterQueue answers "are you +// reachable" without publishing something, the memory backend cannot fail, and +// a queue that is down degrades logging rather than stopping requests - which +// is a reason to alert, not a reason to leave the load balancer pool. +func Ready(ctx context.Context) []Check { + return []Check{ + safely("database", func() error { return pingDB(ctx) }), + safely("cache", probeCache), + } +} + +// safely turns a panic into a failed check. +// +// Not defensive habit: the accessors hand back wrappers, not the resources. +// sdk.Runtime.GetCacheAdapter builds a runtime.Cache around whatever is +// configured and returns it even when nothing is - so the value is not nil, the +// cache inside it is, and the first call dereferences it. A nil check cannot +// see that, and the same is true of GetQueueAdapter. +// +// Whatever the reason, a probe is the last thing that should be able to take +// the process down: the caller is asking whether this instance is well, and +// killing it to answer is the wrong reply. +func safely(name string, fn func() error) (c Check) { + c = Check{Name: name} + defer func() { + if r := recover(); r != nil { + c.OK, c.Err = false, fmt.Sprintf("the check panicked: %v", r) + } + }() + if err := fn(); err != nil { + c.Err = err.Error() + return c + } + c.OK = true + return c +} + +// Healthy reports whether every check passed. +func Healthy(checks []Check) bool { + for _, c := range checks { + if !c.OK { + return false + } + } + return true +} + +func pingDB(ctx context.Context) error { + db := sdk.Runtime.GetDb() + if db == nil { + return errors.New("no database configured") + } + sqlDB, err := db.DB() + if err != nil { + return err + } + return sqlDB.PingContext(ctx) +} + +// cacheProbeKey is written and read back rather than only read: a cache that +// answers "miss" for every key is indistinguishable from a healthy one on a +// read alone, and that is precisely the failure - a client pointed at the wrong +// server - this is meant to catch. +const cacheProbeKey = "go-admin:health" + +func probeCache() error { + adapter := sdk.Runtime.GetCacheAdapter() + if adapter == nil { + return errors.New("no cache configured") + } + want := time.Now().Format(time.RFC3339Nano) + if err := adapter.Set(cacheProbeKey, want, 30); err != nil { + return err + } + got, err := adapter.Get(cacheProbeKey) + if err != nil { + return err + } + if got != want { + return errors.New("the cache returned a different value than was written") + } + return nil +} diff --git a/common/health/health_test.go b/common/health/health_test.go new file mode 100644 index 00000000..c3d7a64c --- /dev/null +++ b/common/health/health_test.go @@ -0,0 +1,141 @@ +package health + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/go-admin-team/go-admin-core/v2/sdk" + "github.com/go-admin-team/go-admin-core/v2/sdk/runtime" + corestorage "github.com/go-admin-team/go-admin-core/v2/storage" +) + +func freshRuntime(t *testing.T) { + t.Helper() + previous := sdk.Runtime + t.Cleanup(func() { sdk.Runtime = previous }) + sdk.Runtime = runtime.NewConfig() +} + +// fakeCache answers whatever the test needs it to. +type fakeCache struct { + setErr error + getErr error + getBack string // returned instead of what was written, when non-empty + stored string +} + +func (c *fakeCache) String() string { return "fake" } +func (c *fakeCache) Set(_ string, val interface{}, _ int) error { + if c.setErr != nil { + return c.setErr + } + c.stored, _ = val.(string) + return nil +} +func (c *fakeCache) Get(string) (string, error) { + if c.getErr != nil { + return "", c.getErr + } + if c.getBack != "" { + return c.getBack, nil + } + return c.stored, nil +} +func (c *fakeCache) Del(string) error { return nil } +func (c *fakeCache) HashGet(_, _ string) (string, error) { return "", nil } +func (c *fakeCache) HashDel(_, _ string) error { return nil } +func (c *fakeCache) Increase(string) error { return nil } +func (c *fakeCache) Decrease(string) error { return nil } +func (c *fakeCache) Expire(string, time.Duration) error { return nil } + +var _ corestorage.AdapterCache = (*fakeCache)(nil) + +func named(checks []Check, name string) Check { + for _, c := range checks { + if c.Name == name { + return c + } + } + return Check{Name: name, Err: "check not reported at all"} +} + +// A cache that accepts writes and answers every read with a different value is +// the failure this probe exists for - a client pointed at the wrong server, or +// one that silently drops everything. A read alone cannot tell that apart from +// a healthy cache with a cold key, which is why the probe writes first. +func TestCacheProbeFailsWhenTheValueDoesNotComeBack(t *testing.T) { + freshRuntime(t) + sdk.Runtime.SetCacheAdapter(&fakeCache{getBack: "something else"}) + + got := named(Ready(context.Background()), "cache") + if got.OK { + t.Error("the cache check passed although the value written was not the value read back") + } + if got.Err == "" { + t.Error("the failing check reported no reason") + } +} + +func TestCacheProbePassesWhenTheValueComesBack(t *testing.T) { + freshRuntime(t) + sdk.Runtime.SetCacheAdapter(&fakeCache{}) + + if got := named(Ready(context.Background()), "cache"); !got.OK { + t.Errorf("the cache check failed for a cache that works: %s", got.Err) + } +} + +func TestCacheProbeReportsAWriteFailure(t *testing.T) { + freshRuntime(t) + sdk.Runtime.SetCacheAdapter(&fakeCache{setErr: errors.New("connection refused")}) + + got := named(Ready(context.Background()), "cache") + if got.OK { + t.Error("the cache check passed although the write failed") + } +} + +// Nothing configured is the case that used to take the process down rather +// than answer. GetCacheAdapter builds a wrapper around whatever is configured +// and returns it even when nothing is, so the value is not nil, the cache +// inside it is, and Set dereferences it - a probe that panics is the worst +// possible answer to "are you well". +// +// Every check has to be reported, passing or not. A probe that omits what it +// could not reach reads as a shorter list of healthy things. +func TestEveryDependencyIsReportedEvenWithNothingConfigured(t *testing.T) { + freshRuntime(t) + + checks := Ready(context.Background()) + for _, name := range []string{"database", "cache"} { + c := named(checks, name) + if c.Err == "check not reported at all" { + t.Errorf("%s was not reported", name) + } + if c.OK { + t.Errorf("%s passed with nothing configured", name) + } + } + if Healthy(checks) { + t.Error("Healthy said yes for a process with no database and no cache") + } +} + +// Draining is what makes the shutdown graceful from the outside: it has to be +// observable before the server stops accepting, or the load balancer learns +// about the shutdown by having its connections cut. +func TestDrainingIsObservableOnceItBegins(t *testing.T) { + previous := draining.Load() + t.Cleanup(func() { draining.Store(previous) }) + + draining.Store(false) + if Draining() { + t.Fatal("Draining reported true before shutdown began") + } + BeginDraining() + if !Draining() { + t.Error("Draining still reported false after BeginDraining") + } +} diff --git a/common/middleware/settings.go b/common/middleware/settings.go index 248a640f..e1eac74e 100644 --- a/common/middleware/settings.go +++ b/common/middleware/settings.go @@ -34,6 +34,7 @@ var CasbinExclude = []UrlInfo{ {Url: "/api/v1/user/pwd", Method: "PUT"}, {Url: "/api/v1/metrics", Method: "GET"}, {Url: "/api/v1/health", Method: "GET"}, + {Url: "/api/v1/ready", Method: "GET"}, {Url: "/", Method: "GET"}, {Url: "/api/v1/server-monitor", Method: "GET"}, {Url: "/api/v1/public/uploadFile", Method: "POST"}, From c579c5f84c2eb70f27fd7b92828d71991e4e3d98 Mon Sep 17 00:00:00 2001 From: zhangwenjian Date: Sun, 6 Sep 2026 09:55:12 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix=F0=9F=90=9B:=20give=20every=20cache=20p?= =?UTF-8?q?robe=20its=20own=20key?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One fixed key meant two probes overlapping - two /ready requests, or two instances against the one redis, which is the normal deployment - overwrote each other's value between the write and the read, and each concluded the cache was broken. A readiness probe that reports false negatives under load pulls healthy instances out of the pool, which is worse than not probing. The key now carries eight random bytes. The written value is still read back, because that is what tells a healthy cache apart from one that answers "miss" for everything, and the key is deleted afterwards on a best-effort basis - its error is dropped deliberately, since the verdict is already decided and a cache that cannot delete what it just wrote is not a reason to refuse traffic. The first version of the concurrency test had no teeth: the probes are short enough that the scheduler ran them one after another, so the fixed-key counter-proof passed three times out of three. The fake cache now holds every writer until all of them have written, which makes the interleaving the test is about actually happen. With one key that is a deterministic 15 failures out of 16 - only the last writer's value survives - and with a key per probe, none. --- common/health/health.go | 38 +++++++++--- common/health/health_test.go | 112 +++++++++++++++++++++++++++++++++-- 2 files changed, 137 insertions(+), 13 deletions(-) diff --git a/common/health/health.go b/common/health/health.go index 79c8ec01..c14ed20b 100644 --- a/common/health/health.go +++ b/common/health/health.go @@ -17,6 +17,8 @@ package health import ( "context" + "crypto/rand" + "encoding/hex" "errors" "fmt" "sync/atomic" @@ -107,22 +109,44 @@ func pingDB(ctx context.Context) error { return sqlDB.PingContext(ctx) } -// cacheProbeKey is written and read back rather than only read: a cache that -// answers "miss" for every key is indistinguishable from a healthy one on a -// read alone, and that is precisely the failure - a client pointed at the wrong -// server - this is meant to catch. -const cacheProbeKey = "go-admin:health" +// cacheProbePrefix names the probe's keys. The key itself is per probe, not +// fixed: two /ready requests arriving together - or two instances sharing one +// redis, which is the normal deployment - would otherwise overwrite each +// other's value between the write and the read and each conclude the cache was +// broken. A readiness probe that reports false negatives under load takes +// healthy instances out of the pool, which is worse than not probing. +const cacheProbePrefix = "go-admin:health:" + +// cacheProbeTTL is short because these keys are write-once and never read +// again by anyone else; it only has to outlive the read that follows. +const cacheProbeTTL = 30 func probeCache() error { adapter := sdk.Runtime.GetCacheAdapter() if adapter == nil { return errors.New("no cache configured") } + + suffix := make([]byte, 8) + if _, err := rand.Read(suffix); err != nil { + return fmt.Errorf("could not build a probe key: %w", err) + } + key := cacheProbePrefix + hex.EncodeToString(suffix) + + // Written and read back rather than only read: a cache that answers "miss" + // for every key - a client pointed at the wrong server - is + // indistinguishable from a healthy one on a read alone. want := time.Now().Format(time.RFC3339Nano) - if err := adapter.Set(cacheProbeKey, want, 30); err != nil { + if err := adapter.Set(key, want, cacheProbeTTL); err != nil { return err } - got, err := adapter.Get(cacheProbeKey) + // Best effort, and its error is deliberately dropped: the verdict is + // already decided by the read below, and a cache that cannot delete a key + // it just wrote is not a reason to refuse traffic. The TTL is the real + // cleanup. + defer func() { _ = adapter.Del(key) }() + + got, err := adapter.Get(key) if err != nil { return err } diff --git a/common/health/health_test.go b/common/health/health_test.go index c3d7a64c..dcb9f717 100644 --- a/common/health/health_test.go +++ b/common/health/health_test.go @@ -3,6 +3,7 @@ package health import ( "context" "errors" + "sync" "testing" "time" @@ -20,30 +21,93 @@ func freshRuntime(t *testing.T) { // fakeCache answers whatever the test needs it to. type fakeCache struct { + mu sync.Mutex setErr error getErr error getBack string // returned instead of what was written, when non-empty - stored string + stored map[string]string + + // oneSlot makes the cache keep a single value however many keys are + // written, which is what a shared probe key turns any cache into. + oneSlot bool + slot string + + // setBarrier, when set, holds every writer until all of them have written. + // Without it the probes are short enough that the scheduler usually runs + // them one after another, and a shared key survives by luck rather than by + // design - which would leave the test below asserting nothing. + setBarrier *barrier +} + +// barrier releases every waiter once n of them have arrived. +type barrier struct { + n int + mu sync.Mutex + got int + ch chan struct{} +} + +func newBarrier(n int) *barrier { return &barrier{n: n, ch: make(chan struct{})} } + +func (b *barrier) wait() { + b.mu.Lock() + b.got++ + if b.got == b.n { + close(b.ch) + } + b.mu.Unlock() + <-b.ch } func (c *fakeCache) String() string { return "fake" } -func (c *fakeCache) Set(_ string, val interface{}, _ int) error { + +func (c *fakeCache) Set(key string, val interface{}, _ int) error { + c.mu.Lock() + defer c.mu.Unlock() if c.setErr != nil { return c.setErr } - c.stored, _ = val.(string) + v, _ := val.(string) + if c.oneSlot { + c.slot = v + return nil + } + if c.stored == nil { + c.stored = map[string]string{} + } + c.stored[key] = v + c.mu.Unlock() + if c.setBarrier != nil { + // Outside the lock on purpose: waiting while holding it would deadlock + // every other writer before the barrier could fill. + c.setBarrier.wait() + } + c.mu.Lock() return nil } -func (c *fakeCache) Get(string) (string, error) { + +func (c *fakeCache) Get(key string) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() if c.getErr != nil { return "", c.getErr } if c.getBack != "" { return c.getBack, nil } - return c.stored, nil + if c.oneSlot { + return c.slot, nil + } + return c.stored[key], nil } -func (c *fakeCache) Del(string) error { return nil } + +func (c *fakeCache) Del(key string) error { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.stored, key) + return nil +} + func (c *fakeCache) HashGet(_, _ string) (string, error) { return "", nil } func (c *fakeCache) HashDel(_, _ string) error { return nil } func (c *fakeCache) Increase(string) error { return nil } @@ -139,3 +203,39 @@ func TestDrainingIsObservableOnceItBegins(t *testing.T) { t.Error("Draining still reported false after BeginDraining") } } + +// Two probes at once must both pass. With one fixed key they overwrite each +// other's value between the write and the read, and a readiness probe that +// reports false negatives under load takes healthy instances out of the pool - +// which is worse than not probing at all. +func TestConcurrentProbesDoNotOverwriteEachOther(t *testing.T) { + freshRuntime(t) + const probes = 16 + sdk.Runtime.SetCacheAdapter(&fakeCache{setBarrier: newBarrier(probes)}) + + var wg sync.WaitGroup + failures := make(chan string, probes) + for i := 0; i < probes; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if c := named(Ready(context.Background()), "cache"); !c.OK { + failures <- c.Err + } + }() + } + wg.Wait() + close(failures) + + var n int + var first string + for err := range failures { + if n == 0 { + first = err + } + n++ + } + if n > 0 { + t.Errorf("%d of %d concurrent probes called a healthy cache broken; first: %s", n, probes, first) + } +}