From 351836b5e4dcfc24bb8cca670c583e4f17bc579c Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 7 Aug 2026 16:51:25 +0800 Subject: [PATCH] feat: honor admin upstream state in health scheduling --- README.md | 8 +- docs/configuration/reference.md | 7 +- internal/controller/bootstrap/bootstrap.go | 15 +- .../controller/health/scheduler_runner.go | 105 +++++++++++++- .../health/scheduler_runner_test.go | 129 ++++++++++++++++++ progress.md | 3 + task_plan.md | 3 +- 7 files changed, 255 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 68e7ebf..92e4051 100644 --- a/README.md +++ b/README.md @@ -221,9 +221,11 @@ Checker 的参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、 `PROXY_POOL_CHECKER_MAX_IN_FLIGHT` 提供。它不会访问 Redis/PostgreSQL；生产 Controller 在启用控制面时装配 Redis 共享任务 broker，并按启用的 Upstream 调度 HTTP/HTTPS/SOCKS5 BASIC 检查、按每个 `check.urls` 创建 EGRESS 任务，并按启用 Routing 的 -`check.targets` 创建 TARGET 任务。调度监督器每轮读取已发布配置，因此 reload 后的上游/路由启停、 -检查间隔、抖动、超时、重试次数、`maxInFlight`、EGRESS URL 和 TARGET Profile 都会在下一轮生效； -BASIC、EGRESS 与 TARGET 以有界轮转组共享上游并发上限。新启用的上游无需重启 Controller。 +`check.targets` 创建 TARGET 任务。调度监督器每轮读取已发布配置；启用 Admin 时只调度配置与 +PostgreSQL 管理态同 revision 且均启用的 Upstream。管理态停用会在下一轮阻止新的 BASIC、EGRESS、 +TARGET 任务，revision 不一致或状态不完整时按失败关闭。因此 reload 后的上游/路由启停、检查间隔、 +抖动、超时、重试次数、`maxInFlight`、EGRESS URL 和 TARGET Profile 都会在下一轮生效；BASIC、 +EGRESS 与 TARGET 以有界轮转组共享上游并发上限。新启用的上游无需重启 Controller。 EGRESS 对成功响应提取纯文本 IP 或常见 JSON IP 字段，并将其作为全局健康事实回传；TARGET 事实 仅归并到对应的 `(routing_name, target_url)` Profile，不改变 Proxy 全局健康。 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index ae04d7b..8fdd999 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -316,7 +316,9 @@ Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-pla 模式只接受回环 Controller 地址。Checker 只从 gRPC 领取任务并批量上报事实，不读取 Redis/PostgreSQL；Controller 在生产启动拓扑中装配 Redis 共享任务队列，当前调度 HTTP/HTTPS/SOCKS5 BASIC、EGRESS 和 TARGET 检查。调度监督器在每轮从已发布配置读取启用的 -上游与 Routing；Admin reload 发布后，上游/路由启停、有效 `check` 策略和目标列表会在下一轮生效， +上游与 Routing；启用 Admin 时还要求 PostgreSQL 管理态与配置 revision 一致，并取两者均启用的 +上游。管理态禁用的上游不会再产生新的 BASIC、EGRESS 或 TARGET 任务；revision 不一致或上游状态 +不完整时本轮失败关闭。Admin reload 发布后，上游/路由启停、有效 `check` 策略和目标列表会在下一轮生效， 新启用的上游无需重启 Controller。 `maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway @@ -648,6 +650,9 @@ proxyAuth: - 启用 Routing 的 `check.targets` 会为其引用的每个启用 Upstream 创建 TARGET 检查组。BASIC、 EGRESS 与 TARGET 使用固定批次和轮转顺序，并共享该 Upstream 的 `check.maxInFlight`，避免 配置多个目标后产生无界检查流量。 +- 启用 Admin 时，调度仅使用配置和管理态均启用、且 revision 一致的 Upstream；管理态停用会在 + 下一调度轮阻止新的 BASIC、EGRESS、TARGET 任务。revision 不一致、重复或缺失的上游状态按失败 + 关闭处理，不读取 Redis due-index。 - 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入 UNHEALTHY。 - `unhealthyRemoveAfter` 控制 UNHEALTHY 持续多久后可由 Controller 回收；`0s` diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index f5faf37..399116d 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -370,7 +370,14 @@ func runWithWorkerFactory( } runners = append(runners, runner) if tasks, ok := opened.activity.(healthTaskRuntime); ok && !nilInterface(tasks) { - schedulers, schedulerErr := newHealthSchedulers(configurationStore, tasks, options.Now) + var schedulerStates []controllerHealth.SchedulerStateSource + if loaded.Value.Admin.Enabled { + if nilInterface(opened.state) { + return errors.Join(ErrStartup, ErrInvalidOptions) + } + schedulerStates = append(schedulerStates, opened.state) + } + schedulers, schedulerErr := newHealthSchedulers(configurationStore, tasks, options.Now, schedulerStates...) if schedulerErr != nil { return fmt.Errorf("%w: build Checker health schedulers: %w", ErrStartup, schedulerErr) } @@ -429,8 +436,10 @@ func newHealthSchedulers( configuration controllerHealth.ConfigurationSource, tasks healthTaskRuntime, now func() time.Time, + states ...controllerHealth.SchedulerStateSource, ) ([]lifecycle.Runner, error) { - if nilInterface(configuration) || nilInterface(tasks) || now == nil { + if nilInterface(configuration) || nilInterface(tasks) || now == nil || len(states) > 1 || + (len(states) == 1 && nilInterface(states[0])) { return nil, ErrInvalidOptions } current := configuration.Current() @@ -449,7 +458,7 @@ func newHealthSchedulers( } supervisor, err := controllerHealth.NewConfiguredSchedulerSupervisor(configuration, tasks, tasks, controllerHealth.SchedulerRunnerOptions{ PollInterval: checkSchedulerPollInterval, BatchSize: checkSchedulerBatchSize, Now: now, - }) + }, states...) if err != nil { return nil, err } diff --git a/internal/controller/health/scheduler_runner.go b/internal/controller/health/scheduler_runner.go index 2fb92e2..f6a396c 100644 --- a/internal/controller/health/scheduler_runner.go +++ b/internal/controller/health/scheduler_runner.go @@ -9,6 +9,7 @@ import ( "time" "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" healthDomain "proxy-pool/internal/domain/health" ) @@ -169,12 +170,26 @@ type ConfiguredSchedulerRunner struct { // bounded BASIC scheduling without a Controller restart. type ConfiguredSchedulerSupervisor struct { configuration ConfigurationSource + state SchedulerStateSource source UpstreamTaskSource sink TaskSink options SchedulerRunnerOptions cursor atomic.Uint64 } +// SchedulerConfigurationSource returns a configuration and its Admin revision +// from one atomic publication. config.Store implements this interface. +type SchedulerConfigurationSource interface { + ConfigurationSource + Snapshot() (*config.Config, uint64) +} + +// SchedulerStateSource provides the persisted Admin view that fences a +// scheduler from combining two different configuration revisions. +type SchedulerStateSource interface { + Snapshot(context.Context) (adminstate.Snapshot, error) +} + func NewConfiguredSchedulerRunner( configuration ConfigurationSource, upstreamID string, @@ -196,14 +211,22 @@ func NewConfiguredSchedulerSupervisor( source UpstreamTaskSource, sink TaskSink, options SchedulerRunnerOptions, + states ...SchedulerStateSource, ) (*ConfiguredSchedulerSupervisor, error) { if nilInterface(configuration) || nilInterface(source) || nilInterface(sink) || options.PollInterval <= 0 || - options.BatchSize <= 0 || options.Now == nil { + options.BatchSize <= 0 || options.Now == nil || len(states) > 1 || (len(states) == 1 && nilInterface(states[0])) { return nil, ErrInvalidSchedulerRunner } - return &ConfiguredSchedulerSupervisor{ + supervisor := &ConfiguredSchedulerSupervisor{ configuration: configuration, source: source, sink: sink, options: options, - }, nil + } + if len(states) == 1 { + if _, ok := configuration.(SchedulerConfigurationSource); !ok { + return nil, ErrInvalidSchedulerRunner + } + supervisor.state = states[0] + } + return supervisor, nil } type TickResult struct { @@ -245,11 +268,10 @@ func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (Tick if err := ctx.Err(); err != nil { return TickResult{}, err } - configuration := supervisor.configuration.Current() - if configuration == nil { - return TickResult{}, ErrInvalidSchedulerRunner + configuration, names, err := supervisor.effectiveConfiguration(ctx) + if err != nil { + return TickResult{}, err } - names := enabledUpstreamNames(configuration) result := TickResult{} for _, name := range names { planner, configured, err := configuredPlannerForConfig(configuration, name) @@ -289,6 +311,54 @@ func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (Tick return result, nil } +// effectiveConfiguration returns the enabled upstream set for one scheduler +// tick. When Admin state is present, it requires the configuration and state +// snapshots to share a revision; an incomplete or torn view schedules nothing. +func (supervisor *ConfiguredSchedulerSupervisor) effectiveConfiguration( + ctx context.Context, +) (*config.Config, []string, error) { + if supervisor == nil || nilInterface(supervisor.configuration) { + return nil, nil, ErrInvalidSchedulerRunner + } + if supervisor.state == nil { + configuration := supervisor.configuration.Current() + if configuration == nil { + return nil, nil, ErrInvalidSchedulerRunner + } + return configuration, enabledUpstreamNames(configuration), nil + } + configurationSource, ok := supervisor.configuration.(SchedulerConfigurationSource) + if !ok || nilInterface(configurationSource) || nilInterface(supervisor.state) { + return nil, nil, ErrInvalidSchedulerRunner + } + configuration, revision := configurationSource.Snapshot() + if configuration == nil || revision == 0 { + return nil, nil, ErrInvalidSchedulerRunner + } + snapshot, err := supervisor.state.Snapshot(ctx) + if err != nil { + return nil, nil, err + } + if snapshot.Config == nil || snapshot.Config.Revision != revision { + return nil, nil, ErrInvalidSchedulerRunner + } + states := make(map[string]adminstate.UpstreamState, len(snapshot.Upstreams)) + for _, upstream := range snapshot.Upstreams { + if upstream.Name == "" || upstream.Revision == 0 { + return nil, nil, ErrInvalidSchedulerRunner + } + if _, duplicate := states[upstream.Name]; duplicate { + return nil, nil, ErrInvalidSchedulerRunner + } + states[upstream.Name] = upstream + } + names, err := enabledUpstreamNamesForState(configuration, states) + if err != nil { + return nil, nil, err + } + return configuration, names, nil +} + func (supervisor *ConfiguredSchedulerSupervisor) upstreamDueGroups( basic DueSource, upstreamID string, @@ -491,6 +561,27 @@ func enabledUpstreamNames(configuration *config.Config) []string { return names } +func enabledUpstreamNamesForState( + configuration *config.Config, + states map[string]adminstate.UpstreamState, +) ([]string, error) { + if configuration == nil { + return nil, ErrInvalidSchedulerRunner + } + names := make([]string, 0, len(configuration.Upstreams)) + for name, upstream := range configuration.Upstreams { + state, exists := states[name] + if !exists { + return nil, ErrInvalidSchedulerRunner + } + if upstream.Enabled && state.Enabled { + names = append(names, name) + } + } + sort.Strings(names) + return names, nil +} + func runScheduler(ctx context.Context, pollInterval time.Duration, tick func(context.Context) (TickResult, error)) error { if ctx == nil || pollInterval <= 0 || tick == nil { return ErrInvalidSchedulerRunner diff --git a/internal/controller/health/scheduler_runner_test.go b/internal/controller/health/scheduler_runner_test.go index f0fad9d..c224231 100644 --- a/internal/controller/health/scheduler_runner_test.go +++ b/internal/controller/health/scheduler_runner_test.go @@ -7,6 +7,7 @@ import ( "time" "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" healthDomain "proxy-pool/internal/domain/health" proxyDomain "proxy-pool/internal/domain/proxy" ) @@ -164,6 +165,102 @@ func TestConfiguredSchedulerSupervisorDiscoversNewEnabledUpstream(t *testing.T) } } +func TestConfiguredSchedulerSupervisorSkipsAdminDisabledUpstream(t *testing.T) { + now := time.Date(2026, 8, 7, 10, 0, 0, 0, time.UTC) + configuration := schedulerConfiguration(now, "provider-a", "provider-b") + configuration.Defaults.Check.URLs = []string{"https://egress.example/identity"} + configuration.Routing = []config.Routing{{ + Name: "checkout", Enabled: true, Upstreams: []string{"provider-a"}, + Check: config.RoutingCheck{Targets: []string{"https://checkout.example/health"}}, + }} + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": { + candidates: []Candidate{{ProxyID: "proxy-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}}, + egressCandidates: map[string][]Candidate{"https://egress.example/identity": { + {ProxyID: "egress-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelEgress, TargetURL: "https://egress.example/identity", DueAt: now}, + }}, + targetCandidates: map[string][]Candidate{targetCandidateKey("checkout", "https://checkout.example/health"): { + {ProxyID: "target-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelTarget, RoutingName: "checkout", TargetURL: "https://checkout.example/health", DueAt: now}, + }}, + }, + "provider-b": {candidates: []Candidate{{ProxyID: "proxy-b", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}}}, + }} + state := schedulerStateStub{snapshot: adminstate.Snapshot{ + Config: &adminstate.ConfigRevision{Revision: 42}, + Upstreams: []adminstate.UpstreamState{ + {Name: "provider-a", Enabled: false, Revision: 11}, + {Name: "provider-b", Enabled: true, Revision: 12}, + }, + }} + sink := &taskSinkStub{} + supervisor, err := NewConfiguredSchedulerSupervisor( + &versionedConfigurationSourceStub{configuration: configuration, revision: 42}, source, sink, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 16, Now: func() time.Time { return now }}, state, + ) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + result, err := supervisor.Tick(context.Background()) + if err != nil || result != (TickResult{Planned: 1, Offered: 1}) || source.sources["provider-a"].inFlightCalls != 0 || + source.sources["provider-a"].dueCalls != 0 || source.sources["provider-a"].egressCalls != 0 || + source.sources["provider-a"].targetCalls != 0 || source.sources["provider-b"].dueCalls != 1 || + len(sink.tasks) != 1 || sink.tasks[0].Candidate.ProxyID != "proxy-b" { + t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + } +} + +func TestConfiguredSchedulerSupervisorFailsClosedOnAdminRevisionMismatch(t *testing.T) { + now := time.Date(2026, 8, 7, 10, 30, 0, 0, time.UTC) + configuration := schedulerConfiguration(now, "provider-a") + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": {candidates: []Candidate{{ProxyID: "proxy-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}}}, + }} + supervisor, err := NewConfiguredSchedulerSupervisor( + &versionedConfigurationSourceStub{configuration: configuration, revision: 42}, source, &taskSinkStub{}, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 16, Now: func() time.Time { return now }}, + schedulerStateStub{snapshot: adminstate.Snapshot{Config: &adminstate.ConfigRevision{Revision: 41}}}, + ) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + if result, err := supervisor.Tick(context.Background()); !errors.Is(err, ErrInvalidSchedulerRunner) || result != (TickResult{}) || + source.sources["provider-a"].inFlightCalls != 0 || source.sources["provider-a"].dueCalls != 0 { + t.Fatalf("Tick() = (%+v, %v); source=%+v", result, err, source.sources["provider-a"]) + } +} + +func TestConfiguredSchedulerSupervisorFailsClosedOnIncompleteAdminState(t *testing.T) { + now := time.Date(2026, 8, 7, 10, 45, 0, 0, time.UTC) + configuration := schedulerConfiguration(now, "provider-a") + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": {candidates: []Candidate{{ProxyID: "proxy-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}}}, + }} + supervisor, err := NewConfiguredSchedulerSupervisor( + &versionedConfigurationSourceStub{configuration: configuration, revision: 42}, source, &taskSinkStub{}, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 16, Now: func() time.Time { return now }}, + schedulerStateStub{snapshot: adminstate.Snapshot{Config: &adminstate.ConfigRevision{Revision: 42}}}, + ) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + if result, err := supervisor.Tick(context.Background()); !errors.Is(err, ErrInvalidSchedulerRunner) || result != (TickResult{}) || + source.sources["provider-a"].inFlightCalls != 0 || source.sources["provider-a"].dueCalls != 0 { + t.Fatalf("Tick() = (%+v, %v); source=%+v", result, err, source.sources["provider-a"]) + } +} + +func TestConfiguredSchedulerSupervisorRequiresVersionedConfigurationWithAdminState(t *testing.T) { + now := time.Date(2026, 8, 7, 11, 0, 0, 0, time.UTC) + supervisor, err := NewConfiguredSchedulerSupervisor( + &configurationSourceStub{configuration: schedulerConfiguration(now, "provider-a")}, + &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{"provider-a": {}}}, &taskSinkStub{}, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 1, Now: func() time.Time { return now }}, schedulerStateStub{}, + ) + if !errors.Is(err, ErrInvalidSchedulerRunner) || supervisor != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor(non-versioned configuration) = (%v, %v)", supervisor, err) + } +} + func TestConfiguredSchedulerSupervisorSchedulesBoundedEgressGroups(t *testing.T) { now := time.Date(2026, 8, 2, 11, 0, 0, 0, time.UTC) configuration := &config.Config{ @@ -310,6 +407,38 @@ func TestConfiguredSchedulerSupervisorNeverExceedsTotalBatchAcrossGroups(t *test } } +func schedulerConfiguration(now time.Time, upstreams ...string) *config.Config { + configured := &config.Config{ + Defaults: config.Defaults{Check: config.Check{ + Interval: config.Duration(time.Minute), MaxInFlight: 2, Timeout: config.Duration(time.Second), MaxAttempts: 1, + }}, + Upstreams: make(map[string]config.Upstream, len(upstreams)), + } + for _, upstream := range upstreams { + configured.Upstreams[upstream] = config.Upstream{Enabled: true} + } + return configured +} + +type versionedConfigurationSourceStub struct { + configuration *config.Config + revision uint64 +} + +func (source *versionedConfigurationSourceStub) Current() *config.Config { + return source.configuration +} + +func (source *versionedConfigurationSourceStub) Snapshot() (*config.Config, uint64) { + return source.configuration, source.revision +} + +type schedulerStateStub struct{ snapshot adminstate.Snapshot } + +func (source schedulerStateStub) Snapshot(context.Context) (adminstate.Snapshot, error) { + return source.snapshot, nil +} + type dueSourceStub struct { inFlight int candidates []Candidate diff --git a/progress.md b/progress.md index ed85ee0..9f39abd 100644 --- a/progress.md +++ b/progress.md @@ -14,6 +14,9 @@ - 管理态禁用当前 Sequential Upstream 现不依赖 Provider 空结果：Controller 以 `ExpectedCurrent` CAS 推进到后续启用项，`loop` 可回绕，`stop`/`stayLast` 无后继时 原子停用路由并刷新完整 Snapshot。定向测试覆盖无 Provider 读取、末端、回绕和跨副本竞争。 +- Health Scheduler 已与 PostgreSQL Admin 管理态对齐：每轮以同一配置 revision 合并启用状态， + 被管理态停用的 Upstream 不再读取 Redis due-index，也不会创建 BASIC、EGRESS、TARGET 任务； + revision 不匹配、状态缺失或重复时失败关闭。 - 全仓 `go test -count=1 -timeout 60s ./...`、`go vet ./...`、`go build ./...`、 Protobuf descriptor、Kustomize Base 渲染及开发证书 SAN/SPIFFE 校验均通过。Compose 容器端到端启动在拉取 Dockerfile 前端与监控镜像时受 Docker Desktop HTTPS 代理缺失阻断， diff --git a/task_plan.md b/task_plan.md index 706e2c7..1fb15c6 100644 --- a/task_plan.md +++ b/task_plan.md @@ -47,7 +47,8 @@ Redis 原子复核策略、归属与 assignment epoch；两类自动 Drain 均已接入低基数 Prometheus 指标。Admin 成功提交的 Upstream 启停、Routing 切换和配置发布现会经 公用广播器立即刷新本进程所有 Worker 完整 Snapshot；跨 Controller 副本仍以定时刷新 - 收敛。后续补齐更完整的运行态可观测闭环。 + 收敛。Health Scheduler 现以相同 revision 合并配置与管理态启用状态，管理态停用会停止 + 该 Upstream 的新 BASIC/EGRESS/TARGET 任务。后续补齐更完整的运行态可观测闭环。 ## 串并行关系