From a557ba5af0c17fed2e46b287f5d3d374386ce081 Mon Sep 17 00:00:00 2001 From: k2safe Date: Sun, 20 Sep 2026 20:07:32 +0800 Subject: [PATCH] feat: configure graceful shutdown window --- .env.example | 2 ++ cmd/orbit/main.go | 9 +++++---- docs/20-operations-runbook.md | 4 ++-- docs/21-container-deployment.md | 1 + docs/api/observability.md | 1 + docs/reports/phase-7-regression-status.md | 4 ++-- internal/config/config.go | 9 +++++++++ internal/config/config_test.go | 6 ++++-- internal/observability/catalog.go | 1 + internal/observability/metrics_test.go | 4 ++-- internal/observability/runtime.go | 4 +++- 11 files changed, 32 insertions(+), 13 deletions(-) diff --git a/.env.example b/.env.example index 4a9e305..7083d03 100644 --- a/.env.example +++ b/.env.example @@ -10,6 +10,8 @@ ORBIT_SCHEDULE_TICK_INTERVAL=1s ORBIT_RECOVERY_TICK_INTERVAL=2s # 每个后台批次最多处理的持久化候选，范围1–100。 ORBIT_RUNTIME_BATCH_LIMIT=100 +# 收到 SIGTERM 后的排空与优雅关闭窗口，范围1s–5m，默认10s。 +ORBIT_SHUTDOWN_TIMEOUT=10s # Empty disables metrics. To enable a separate private listener, use e.g. # ORBIT_METRICS_ADDR=127.0.0.1:9090. Never expose it on the public API ingress. ORBIT_METRICS_ADDR= diff --git a/cmd/orbit/main.go b/cmd/orbit/main.go index bd4a25f..7dfde87 100644 --- a/cmd/orbit/main.go +++ b/cmd/orbit/main.go @@ -74,6 +74,7 @@ func run() error { ScheduleTickInterval: cfg.ScheduleTickInterval, RecoveryTickInterval: cfg.RecoveryTickInterval, BatchLimit: cfg.RuntimeBatchLimit, + ShutdownTimeout: cfg.ShutdownTimeout, }) metrics.RecordDatabase(db.Stats()) } @@ -94,7 +95,7 @@ func run() error { httpserver.RegisterEventSubscriptions(server, eventService) httpserver.RegisterEventConsumers(server, eventHub) defer func() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout) defer cancel() _ = eventHub.Shutdown(ctx) }() @@ -161,7 +162,7 @@ func run() error { }() defer func() { stopSchedules(); <-scheduleDone }() defer func() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout) defer cancel() _ = hub.Shutdown(ctx) }() @@ -249,7 +250,7 @@ func run() error { slog.Info("Orbit metrics server starting", "metrics_address", metricsListener.Addr().String()) } go func() { done <- server.Serve(listener) }() - slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node, "workflow_tick_interval", cfg.WorkflowTickInterval.String(), "schedule_tick_interval", cfg.ScheduleTickInterval.String(), "recovery_tick_interval", cfg.RecoveryTickInterval.String(), "runtime_batch_limit", cfg.RuntimeBatchLimit) + slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node, "workflow_tick_interval", cfg.WorkflowTickInterval.String(), "schedule_tick_interval", cfg.ScheduleTickInterval.String(), "recovery_tick_interval", cfg.RecoveryTickInterval.String(), "runtime_batch_limit", cfg.RuntimeBatchLimit, "shutdown_timeout", cfg.ShutdownTimeout.String()) select { case err := <-done: if errors.Is(err, http.ErrServerClosed) { @@ -258,7 +259,7 @@ func run() error { return err case <-root.Done(): readiness.BeginDrain() - shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) + shutdown, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout) defer cancel() if err := server.Shutdown(shutdown); err != nil { _ = server.Close() diff --git a/docs/20-operations-runbook.md b/docs/20-operations-runbook.md index 5f445ce..b3cef6c 100644 --- a/docs/20-operations-runbook.md +++ b/docs/20-operations-runbook.md @@ -37,11 +37,11 @@ curl -fsS http://127.0.0.1:8080/health/ready `live` 只表示进程仍在运行；`ready` 同时核对 MySQL、迁移账本和零外键要求。收到 `503` 时停止向该节点发送新的受理、执行授权和流程推进请求，保留原幂等键并等待核对，不把网络错误当作业务失败。 -后台调度默认每秒检查 Weave 和计划、每两秒执行一次执行恢复，每批最多处理100条持久化候选。容量验证后可以通过 `ORBIT_WORKFLOW_TICK_INTERVAL`、`ORBIT_SCHEDULE_TICK_INTERVAL`、`ORBIT_RECOVERY_TICK_INTERVAL` 和 `ORBIT_RUNTIME_BATCH_LIMIT` 调整；间隔只允许 `100ms` 到 `1m`，批量只允许 `1` 到 `100`。调优必须和数据库锁等待、连接池、积压及恢复窗口一起记录，不能只提高批量或缩短间隔来掩盖容量不足。 +后台调度默认每秒检查 Weave 和计划、每两秒执行一次执行恢复，每批最多处理100条持久化候选。容量验证后可以通过 `ORBIT_WORKFLOW_TICK_INTERVAL`、`ORBIT_SCHEDULE_TICK_INTERVAL`、`ORBIT_RECOVERY_TICK_INTERVAL` 和 `ORBIT_RUNTIME_BATCH_LIMIT` 调整；间隔只允许 `100ms` 到 `1m`，批量只允许 `1` 到 `100`。`ORBIT_SHUTDOWN_TIMEOUT` 控制 SIGTERM 后的排空和优雅关闭窗口，允许 `1s` 到 `5m`，默认 `10s`。调优必须和数据库锁等待、连接池、积压及恢复窗口一起记录，不能只提高批量或缩短间隔来掩盖容量不足。 ## 3. 滚动停止与替换 -Orbit 接收到 `SIGTERM` 后停止新的 Worker/Event 连接，取消后台循环，关闭 HTTP/WS 监听并在有限时间内等待已有连接退出。发布脚本应按以下顺序操作： +Orbit 接收到 `SIGTERM` 后停止新的 Worker/Event 连接，先将 readiness 置为 `503`，再取消后台循环、关闭 HTTP/WS 监听，并按 `ORBIT_SHUTDOWN_TIMEOUT` 等待已有连接退出。发布脚本应按以下顺序操作： 1. 从服务发现或负载均衡中摘除目标节点，等待健康检查不再分配新流量。 2. 向进程发送 `SIGTERM`，等待进程退出；超过发布窗口时记录进程和连接状态，再按值班策略处理。 diff --git a/docs/21-container-deployment.md b/docs/21-container-deployment.md index 5033d98..f72f0b1 100644 --- a/docs/21-container-deployment.md +++ b/docs/21-container-deployment.md @@ -53,6 +53,7 @@ docker run --rm \ --env ORBIT_WORKFLOW_TICK_INTERVAL=1s \ --env ORBIT_SCHEDULE_TICK_INTERVAL=1s \ --env ORBIT_RECOVERY_TICK_INTERVAL=2s \ + --env ORBIT_SHUTDOWN_TIMEOUT=10s \ --env ORBIT_METRICS_ADDR=127.0.0.1:9090 \ --publish 8080:8080 \ registry.example/orbit:dev diff --git a/docs/api/observability.md b/docs/api/observability.md index 195fcf4..3e5359c 100644 --- a/docs/api/observability.md +++ b/docs/api/observability.md @@ -46,6 +46,7 @@ curl -fsS http://127.0.0.1:9090/metrics | `orbit_runtime_batch_last_success_timestamp_seconds` | gauge | 最后一次无错误批次返回的Unix时间；启动后未成功时为0 | | `orbit_runtime_tick_interval_seconds{component}` | gauge | 节点实际生效的后台扫描间隔，component为workflow、schedule、execution_recovery | | `orbit_runtime_batch_limit` | gauge | 节点实际生效的后台批量上限 | +| `orbit_shutdown_timeout_seconds` | gauge | 节点实际生效的排空与优雅关闭窗口（秒） | | `orbit_db_connections_open` / `orbit_db_connections_in_use` / `orbit_db_connections_idle` | gauge | 本节点MySQL连接池当前连接数 | | `orbit_db_connections_max_open` | gauge | 池配置的最大连接数 | | `orbit_db_wait_count_total` / `orbit_db_wait_duration_seconds_total` | counter | 连接池累计等待次数/秒数，不是数据库锁等待 | diff --git a/docs/reports/phase-7-regression-status.md b/docs/reports/phase-7-regression-status.md index a9bede2..ef636b4 100644 --- a/docs/reports/phase-7-regression-status.md +++ b/docs/reports/phase-7-regression-status.md @@ -8,8 +8,8 @@ | --- | --- | --- | | 就绪探针排空门控 | `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 间隔和批量上限以固定基数指标导出，不进入业务或租户标签 | +| 后台运行参数边界 | `internal/config.TestRuntimeDefaultsAndBoundedOverrides`；默认值与五个 `ORBIT_*` 覆盖项 | PASS；循环间隔限定100ms–1m，批量限定1–100，排空超时限定1s–5m，服务启动日志记录生效值 | +| 生效运行参数指标 | `internal/observability.TestRuntimeObservationDoesNotExposeErrors`；固定指标目录 | PASS；workflow/schedule/execution_recovery 间隔、批量上限和排空超时以固定基数指标导出，不进入业务或租户标签 | | SDK 与独立示例竞态、vet、构建 | Docker `golang:1.27.1`；SDK 全 module、全部示例、`orbit-load`、`orbit-exec-load`、`orbit-schedule-load`、`orbit-workflow-load`，以及 result-inbox/task-outbox | PASS；健康/版本 SDK 契约包含 `Client.Version(ctx)`，返回协议身份且不暴露敏感响应 | | 容器发布物构建 | `make image`；宿主平台 linux/arm64；显式 `docker build --platform linux/amd64`；带 `VERSION`、`REVISION`、`BUILD_TIME` 构建参数的版本化镜像；镜像内 `GOWORK=off` | PASS；native 与 amd64 镜像均构建成功，包含 `orbit`、`orbit-migrate`、`orbit-admin`，运行用户为非 root；版本化 amd64 二进制和 OCI 标签均包含注入的构建身份，`GET /version` 契约单测通过；无 DSN 时服务按预期拒绝启动 | | 当前协议门禁 | `make protocol-check` | PASS，当前 v1 契约同步 | diff --git a/internal/config/config.go b/internal/config/config.go index 2ed38c7..5015905 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -19,6 +19,7 @@ type Config struct { ScheduleTickInterval time.Duration // ScheduleTickInterval 计划扫描间隔，范围100ms–1m。 RecoveryTickInterval time.Duration // RecoveryTickInterval 执行租约恢复扫描间隔，范围100ms–1m。 RuntimeBatchLimit int // RuntimeBatchLimit 每个后台批次的候选上限，范围1–100。 + ShutdownTimeout time.Duration // ShutdownTimeout 优雅排空和关闭窗口，范围1s–5m。 } func Load() (Config, error) { return FromEnv(os.Getenv) } @@ -33,6 +34,7 @@ func FromEnv(get func(string) string) (Config, error) { ScheduleTickInterval: time.Second, RecoveryTickInterval: 2 * time.Second, RuntimeBatchLimit: 100, + ShutdownTimeout: 10 * time.Second, } if c.HTTPAddr == "" { c.HTTPAddr = "127.0.0.1:8080" @@ -77,6 +79,13 @@ func FromEnv(get func(string) string) (Config, error) { } c.RuntimeBatchLimit = n } + if raw := get("ORBIT_SHUTDOWN_TIMEOUT"); raw != "" { + d, err := time.ParseDuration(raw) + if err != nil || d < time.Second || d > 5*time.Minute { + return Config{}, errors.New("ORBIT_SHUTDOWN_TIMEOUT must be a duration from 1s to 5m") + } + c.ShutdownTimeout = d + } if c.MetricsAddr != "" { _, port, err := net.SplitHostPort(c.MetricsAddr) n, parseErr := strconv.Atoi(port) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index f8e3863..57ef155 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -9,15 +9,16 @@ func TestRuntimeDefaultsAndBoundedOverrides(t *testing.T) { base := map[string]string{"ORBIT_MYSQL_DSN": "private-dsn"} get := func(k string) string { return base[k] } c, err := FromEnv(get) - if err != nil || c.WorkflowTickInterval != time.Second || c.ScheduleTickInterval != time.Second || c.RecoveryTickInterval != 2*time.Second || c.RuntimeBatchLimit != 100 { + if err != nil || c.WorkflowTickInterval != time.Second || c.ScheduleTickInterval != time.Second || c.RecoveryTickInterval != 2*time.Second || c.RuntimeBatchLimit != 100 || c.ShutdownTimeout != 10*time.Second { t.Fatalf("unexpected runtime defaults: %+v err=%v", c, err) } base["ORBIT_WORKFLOW_TICK_INTERVAL"] = "250ms" base["ORBIT_SCHEDULE_TICK_INTERVAL"] = "1500ms" base["ORBIT_RECOVERY_TICK_INTERVAL"] = "1m" base["ORBIT_RUNTIME_BATCH_LIMIT"] = "37" + base["ORBIT_SHUTDOWN_TIMEOUT"] = "45s" c, err = FromEnv(get) - if err != nil || c.WorkflowTickInterval != 250*time.Millisecond || c.ScheduleTickInterval != 1500*time.Millisecond || c.RecoveryTickInterval != time.Minute || c.RuntimeBatchLimit != 37 { + if err != nil || c.WorkflowTickInterval != 250*time.Millisecond || c.ScheduleTickInterval != 1500*time.Millisecond || c.RecoveryTickInterval != time.Minute || c.RuntimeBatchLimit != 37 || c.ShutdownTimeout != 45*time.Second { t.Fatalf("valid runtime override rejected: %+v err=%v", c, err) } for key, value := range map[string]string{ @@ -25,6 +26,7 @@ func TestRuntimeDefaultsAndBoundedOverrides(t *testing.T) { "ORBIT_SCHEDULE_TICK_INTERVAL": "2m", "ORBIT_RECOVERY_TICK_INTERVAL": "not-a-duration", "ORBIT_RUNTIME_BATCH_LIMIT": "101", + "ORBIT_SHUTDOWN_TIMEOUT": "500ms", } { base[key] = value if _, err := FromEnv(get); err == nil { diff --git a/internal/observability/catalog.go b/internal/observability/catalog.go index 98d6cbc..372b409 100644 --- a/internal/observability/catalog.go +++ b/internal/observability/catalog.go @@ -64,5 +64,6 @@ func catalog() map[string]descriptor { c["orbit_business_snapshot_truncated"] = descriptor{gauge, "Whether a bounded business-state sample hit its per-state cap", nil} c["orbit_runtime_tick_interval_seconds"] = descriptor{gauge, "Configured background controller tick interval in seconds", map[string][]string{"component": components}} c["orbit_runtime_batch_limit"] = descriptor{gauge, "Configured maximum candidates per background batch", nil} + c["orbit_shutdown_timeout_seconds"] = descriptor{gauge, "Configured graceful drain and shutdown window in seconds", nil} return c } diff --git a/internal/observability/metrics_test.go b/internal/observability/metrics_test.go index e5654b1..b804286 100644 --- a/internal/observability/metrics_test.go +++ b/internal/observability/metrics_test.go @@ -96,7 +96,7 @@ func TestInvalidNumbersAndMonotoneCounters(t *testing.T) { } func TestRuntimeObservationDoesNotExposeErrors(t *testing.T) { r := NewRegistry() - r.RecordRuntimeConfig(RuntimeConfig{WorkflowTickInterval: time.Second, ScheduleTickInterval: 1500 * time.Millisecond, RecoveryTickInterval: 2 * time.Second, BatchLimit: 37}) + r.RecordRuntimeConfig(RuntimeConfig{WorkflowTickInterval: time.Second, ScheduleTickInterval: 1500 * time.Millisecond, RecoveryTickInterval: 2 * time.Second, BatchLimit: 37, ShutdownTimeout: 45 * time.Second}) r.RecordBatch("workflow", time.Now().Add(-time.Millisecond), nil) r.RecordBatch("workflow", time.Now(), errors.New("private-dsn")) r.RecordBatch("schedule", time.Now(), context.Canceled) @@ -107,7 +107,7 @@ func TestRuntimeObservationDoesNotExposeErrors(t *testing.T) { t.Fatal("missing runtime evidence", want) } } - for _, want := range []string{`orbit_runtime_tick_interval_seconds{component="workflow"} 1`, `orbit_runtime_tick_interval_seconds{component="schedule"} 1.5`, `orbit_runtime_tick_interval_seconds{component="execution_recovery"} 2`, "orbit_runtime_batch_limit 37\n"} { + for _, want := range []string{`orbit_runtime_tick_interval_seconds{component="workflow"} 1`, `orbit_runtime_tick_interval_seconds{component="schedule"} 1.5`, `orbit_runtime_tick_interval_seconds{component="execution_recovery"} 2`, "orbit_runtime_batch_limit 37\n", "orbit_shutdown_timeout_seconds 45\n"} { if !strings.Contains(out, want) { t.Fatal("missing runtime configuration", want) } diff --git a/internal/observability/runtime.go b/internal/observability/runtime.go index 8e86b7f..d18ee82 100644 --- a/internal/observability/runtime.go +++ b/internal/observability/runtime.go @@ -17,17 +17,19 @@ type RuntimeConfig struct { ScheduleTickInterval time.Duration RecoveryTickInterval time.Duration BatchLimit int + ShutdownTimeout time.Duration } // RecordRuntimeConfig records the effective bounded controller settings. func (r *Registry) RecordRuntimeConfig(config RuntimeConfig) { - if r == nil || config.WorkflowTickInterval <= 0 || config.ScheduleTickInterval <= 0 || config.RecoveryTickInterval <= 0 || config.BatchLimit <= 0 { + if r == nil || config.WorkflowTickInterval <= 0 || config.ScheduleTickInterval <= 0 || config.RecoveryTickInterval <= 0 || config.BatchLimit <= 0 || config.ShutdownTimeout <= 0 { return } r.SetGauge("orbit_runtime_tick_interval_seconds", config.WorkflowTickInterval.Seconds(), Labels{"component": "workflow"}) r.SetGauge("orbit_runtime_tick_interval_seconds", config.ScheduleTickInterval.Seconds(), Labels{"component": "schedule"}) r.SetGauge("orbit_runtime_tick_interval_seconds", config.RecoveryTickInterval.Seconds(), Labels{"component": "execution_recovery"}) r.SetGauge("orbit_runtime_batch_limit", float64(config.BatchLimit), nil) + r.SetGauge("orbit_shutdown_timeout_seconds", config.ShutdownTimeout.Seconds(), nil) } // initializeProcess is called once by the registry constructor.