From 00d4156b2530ca70e1ee1e085c4b305d08df97de Mon Sep 17 00:00:00 2001 From: k2safe Date: Sun, 20 Sep 2026 19:41:28 +0800 Subject: [PATCH] feat: drain readiness before graceful shutdown --- cmd/orbit/main.go | 15 ++++----- docs/api/health.md | 4 ++- docs/reports/phase-7-regression-status.md | 3 +- internal/transport/httpserver/server.go | 33 ++++++++++++++++++++ internal/transport/httpserver/server_test.go | 20 ++++++++++++ 5 files changed, 66 insertions(+), 9 deletions(-) diff --git a/cmd/orbit/main.go b/cmd/orbit/main.go index fd095b5..bd4a25f 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -54,6 +54,12 @@ func run() error { return err } cancel() + readiness := httpserver.NewReadiness(func(ctx context.Context) error { + if err := db.PingContext(ctx); err != nil { + return err + } + return mysql.CheckSchema(ctx, db) + }) var metrics *observability.Registry var metricsListener net.Listener if cfg.MetricsAddr != "" { @@ -71,13 +77,7 @@ func run() error { }) metrics.RecordDatabase(db.Stats()) } - ready := func(ctx context.Context) error { - if err := db.PingContext(ctx); err != nil { - return err - } - return mysql.CheckSchema(ctx, db) - } - server := httpserver.New(cfg.HTTPAddr, ready, service.New(mysql.NewTaskStore(db))) + server := httpserver.New(cfg.HTTPAddr, readiness.Check, service.New(mysql.NewTaskStore(db))) httpserver.RegisterApplicationResources(server, resources.New(mysql.NewTaskStore(db))) node, err := core.NewID() if err != nil { @@ -257,6 +257,7 @@ func run() error { } return err case <-root.Done(): + readiness.BeginDrain() shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := server.Shutdown(shutdown); err != nil { diff --git a/docs/api/health.md b/docs/api/health.md index c86ac97..dfe0bc6 100644 --- a/docs/api/health.md +++ b/docs/api/health.md @@ -1,6 +1,6 @@ # 健康检查接口 -状态：已实现，基础阶段。无需鉴权；默认仅监听 `127.0.0.1:8080`。 +状态：已实现。无需鉴权；默认仅监听 `127.0.0.1:8080`。 ## GET /health/live @@ -38,6 +38,8 @@ curl -i http://127.0.0.1:8080/health/ready 响应不包含 DSN、数据库错误文本或凭据。检查迁移记录不能发现所有人工 DDL 漂移，运行账户权限和迁移流程仍需受控。 +进程收到 `SIGTERM` 后会先进入排空状态：就绪探针立即返回 `503`，再在有限的优雅关闭窗口内停止 HTTP/WS 服务并等待存量请求结束。存量任务租约不会因为探针变化而被删除，仍由正常恢复逻辑接管。 + ## GET /version 返回当前服务的构建和协议身份，不访问数据库，也无需鉴权。该接口适合在滚动发布或故障接管后核对节点是否运行了预期版本。 diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index c5a8f1b..b6b17ed 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -6,6 +6,7 @@ | 检查 | 环境/命令 | 结果 | | --- | --- | --- | +| 就绪探针排空门控 | `go test -race ./internal/transport/httpserver ./cmd/orbit`；`TestReadinessEntersDrainBeforeShutdown` | PASS；SIGTERM 关闭路径先将 `/health/ready` 置为 503，再进入限定优雅关闭窗口；响应不泄露内部错误 | | 服务端单元与竞态、vet、构建 | Docker `golang:1.27.1`，仓库声明版本；`GOWORK=off go test -race ./... && go vet ./... && go build ./...` | PASS | | 后台运行参数边界 | `internal/config.TestRuntimeDefaultsAndBoundedOverrides`；默认值与四个 `ORBIT_*` 覆盖项 | PASS；循环间隔限定100ms–1m，批量限定1–100，服务启动日志记录生效值 | | 生效运行参数指标 | `internal/observability.TestRuntimeObservationDoesNotExposeErrors`；固定指标目录 | PASS；workflow/schedule/execution_recovery 间隔和批量上限以固定基数指标导出，不进入业务或租户标签 | @@ -28,7 +29,7 @@ | Go 1.27.1 备份恢复脚本 | Go 1.27.1 交叉构建 macOS arm64 二进制；`scripts/test-backup-restore.sh`；MySQL 8.4 | PASS；源/恢复核心计数均为 `1,1,1,1,28`，迁移校验完整、固定索引6个、外键0 | | Go 1.27.1 数据库短暂不可用脚本 | Go 1.27.1 交叉构建 macOS arm64 二进制；`scripts/test-db-recovery.sh`；MySQL 8.4 同卷重启 | PASS；readiness `200 → 503 → 200`，恢复后使用原凭据成功受理任务 | | Go 1.27.1 MySQL GTID 复制与手工提升 | 两个一次性 MySQL 8.4 bridge 容器；真实 Orbit HTTP；`scripts/test-mysql-replication.sh` | PASS；复制 IO/SQL 均为 `Yes`，任务记录复制到副本，源库停止后手工提升、结构核验和原幂等键重放通过；自动切换未验证 | -| Orbit HTTP/metrics 子进程 | 同一共享网络命名空间；`TestIntegrationMetricsProcess`；metrics 启用/禁用各一组，禁用分支等待后台周期 | PASS，5.304 秒；监听隔离、隐私、后台采样、metrics 关闭后调度器持续运行和优雅退出均通过 | +| Orbit HTTP/metrics 子进程 | 同一共享网络命名空间；`TestIntegrationMetricsProcess`；metrics 启用/禁用各一组，禁用分支等待后台周期 | PASS，5.304 秒；监听隔离、隐私、后台采样、metrics 关闭后调度器持续运行和优雅退出均通过；服务就绪排空门控另有单元/竞态覆盖 | | 真实Worker执行测量 | `orbit-exec-load`；最终短/混合/中长及截止不足四组，环境及原始JSON见[测量报告](phase-6-execution-load.md) | 96+96+6项全部逐项核验，长任务18次续租；24项截止负例按预期非零退出。SDK全模块竞态/vet/build通过，生产容量未验收 | | 真实Weave流程测量 | `orbit-workflow-load`；24/100实例、固定发布版本双分支并行及24实例截止负例，见[流程测量报告](phase-6-workflow-load.md) | 124个正常实例、248个不同任务/唯一尝试和本地调用逐项关联；负例2成功/22未核验，非零退出。重新构建并保存源码/二进制哈希，替换旧证据；生产容量未验收 | diff --git a/internal/transport/httpserver/server.go b/internal/transport/httpserver/server.go index 646fc08..d9e23ad 100644 --- a/internal/transport/httpserver/server.go +++ b/internal/transport/httpserver/server.go @@ -4,7 +4,9 @@ package httpserver import ( "context" "encoding/json" + "errors" "net/http" + "sync/atomic" "time" "git.yuntuo.hk/k2safe/orbit/internal/buildinfo" @@ -14,6 +16,37 @@ import ( type Checker func(context.Context) error +// Readiness 将数据库/结构检查与进程排空状态合并为一个就绪门控。 +// 收到终止信号后先调用 BeginDrain，再停止接收新请求；存量请求仍由 +// http.Server.Shutdown 在限定时间内完成。这样编排平台有机会先看到 503。 +type Readiness struct { + checker Checker + draining atomic.Bool +} + +// NewReadiness 创建可在优雅下线期间切换为未就绪的门控。 +func NewReadiness(checker Checker) *Readiness { + return &Readiness{checker: checker} +} + +// Check 执行就绪检查；排空开始后始终返回内部错误，由 HTTP 层统一隐藏细节。 +func (r *Readiness) Check(ctx context.Context) error { + if r == nil || r.draining.Load() { + return errors.New("server is draining") + } + if r.checker == nil { + return errors.New("readiness checker is unavailable") + } + return r.checker(ctx) +} + +// BeginDrain 先于 http.Server.Shutdown 调用，阻止排空窗口内的新流量。 +func (r *Readiness) BeginDrain() { + if r != nil { + r.draining.Store(true) + } +} + func New(addr string, ready Checker, tasks ...*service.Service) *http.Server { mux := http.NewServeMux() write := func(w http.ResponseWriter, status int, body map[string]string) { diff --git a/internal/transport/httpserver/server_test.go b/internal/transport/httpserver/server_test.go index 4387f82..6479b3b 100644 --- a/internal/transport/httpserver/server_test.go +++ b/internal/transport/httpserver/server_test.go @@ -22,6 +22,26 @@ func TestHealthSeparatesLivenessAndReadiness(t *testing.T) { } } +func TestReadinessEntersDrainBeforeShutdown(t *testing.T) { + checks := 0 + readiness := NewReadiness(func(context.Context) error { + checks++ + return nil + }) + server := New("127.0.0.1:0", readiness.Check) + ready := httptest.NewRecorder() + server.Handler.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/health/ready", nil)) + if ready.Code != http.StatusOK || checks != 1 { + t.Fatalf("before drain: status=%d checks=%d", ready.Code, checks) + } + readiness.BeginDrain() + draining := httptest.NewRecorder() + server.Handler.ServeHTTP(draining, httptest.NewRequest(http.MethodGet, "/health/ready", nil)) + if draining.Code != http.StatusServiceUnavailable || checks != 1 || strings.Contains(draining.Body.String(), "server is draining") { + t.Fatalf("during drain: status=%d checks=%d body=%s", draining.Code, checks, draining.Body.String()) + } +} + func TestVersionIsPublicAndBounded(t *testing.T) { server := New("127.0.0.1:0", nil) rr := httptest.NewRecorder()