From a191ca583725c6e6e845a864f5e9d442c2c8a809 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 7 Aug 2026 16:57:30 +0800 Subject: [PATCH] feat: honor admin routing state in target scheduling --- README.md | 5 +- docs/configuration/reference.md | 7 +- .../controller/health/scheduler_runner.go | 89 ++++++++++++++----- .../health/scheduler_runner_test.go | 58 ++++++++++++ progress.md | 2 +- task_plan.md | 3 +- 6 files changed, 134 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 92e4051..a920aa7 100644 --- a/README.md +++ b/README.md @@ -222,8 +222,9 @@ Checker 的参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、 Controller 在启用控制面时装配 Redis 共享任务 broker，并按启用的 Upstream 调度 HTTP/HTTPS/SOCKS5 BASIC 检查、按每个 `check.urls` 创建 EGRESS 任务，并按启用 Routing 的 `check.targets` 创建 TARGET 任务。调度监督器每轮读取已发布配置；启用 Admin 时只调度配置与 -PostgreSQL 管理态同 revision 且均启用的 Upstream。管理态停用会在下一轮阻止新的 BASIC、EGRESS、 -TARGET 任务，revision 不一致或状态不完整时按失败关闭。因此 reload 后的上游/路由启停、检查间隔、 +PostgreSQL 管理态同 revision 且均启用的 Upstream 和 Routing。管理态停用 Upstream 会在下一轮阻止新的 +BASIC、EGRESS、TARGET 任务；管理态停用 Routing 则停止该 Routing 的新 TARGET 任务。revision 不一致或 +状态不完整时按失败关闭。因此 reload 后的上游/路由启停、检查间隔、 抖动、超时、重试次数、`maxInFlight`、EGRESS URL 和 TARGET Profile 都会在下一轮生效；BASIC、 EGRESS 与 TARGET 以有界轮转组共享上游并发上限。新启用的上游无需重启 Controller。 EGRESS 对成功响应提取纯文本 IP 或常见 JSON IP 字段，并将其作为全局健康事实回传；TARGET 事实 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 8fdd999..78e6566 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -317,8 +317,9 @@ Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-pla Redis/PostgreSQL；Controller 在生产启动拓扑中装配 Redis 共享任务队列，当前调度 HTTP/HTTPS/SOCKS5 BASIC、EGRESS 和 TARGET 检查。调度监督器在每轮从已发布配置读取启用的 上游与 Routing；启用 Admin 时还要求 PostgreSQL 管理态与配置 revision 一致，并取两者均启用的 -上游。管理态禁用的上游不会再产生新的 BASIC、EGRESS 或 TARGET 任务；revision 不一致或上游状态 -不完整时本轮失败关闭。Admin reload 发布后，上游/路由启停、有效 `check` 策略和目标列表会在下一轮生效， +上游和 Routing。管理态禁用的上游不会再产生新的 BASIC、EGRESS 或 TARGET 任务；管理态禁用的 +Routing 也不会再产生其 TARGET 任务；revision 不一致或状态不完整时本轮失败关闭。Admin reload 发布后， +上游/路由启停、有效 `check` 策略和目标列表会在下一轮生效， 新启用的上游无需重启 Controller。 `maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway @@ -653,6 +654,8 @@ proxyAuth: - 启用 Admin 时，调度仅使用配置和管理态均启用、且 revision 一致的 Upstream；管理态停用会在 下一调度轮阻止新的 BASIC、EGRESS、TARGET 任务。revision 不一致、重复或缺失的上游状态按失败 关闭处理，不读取 Redis due-index。 +- 启用 Admin 时，TARGET Profile 还要求其 Routing 在配置和管理态中均启用；管理态停用 Routing + 仅停止该 Routing 的新 TARGET 任务，不影响同一 Upstream 的 BASIC/EGRESS 或其他启用 Routing。 - 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入 UNHEALTHY。 - `unhealthyRemoveAfter` 控制 UNHEALTHY 持续多久后可由 Controller 回收；`0s` diff --git a/internal/controller/health/scheduler_runner.go b/internal/controller/health/scheduler_runner.go index f6a396c..4e3ac16 100644 --- a/internal/controller/health/scheduler_runner.go +++ b/internal/controller/health/scheduler_runner.go @@ -268,7 +268,7 @@ func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (Tick if err := ctx.Err(); err != nil { return TickResult{}, err } - configuration, names, err := supervisor.effectiveConfiguration(ctx) + configuration, names, routings, err := supervisor.effectiveConfiguration(ctx) if err != nil { return TickResult{}, err } @@ -286,7 +286,7 @@ func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (Tick return TickResult{}, err } check := config.EffectiveCheck(configuration.Defaults.Check, configuration.Upstreams[name].Check) - groups, err := supervisor.upstreamDueGroups(basicSource, name, check.URLs, configuredTargetProfiles(configuration, name)) + groups, err := supervisor.upstreamDueGroups(basicSource, name, check.URLs, configuredTargetProfiles(configuration, name, routings)) if err != nil { return TickResult{}, err } @@ -311,52 +311,89 @@ 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. +// effectiveConfiguration returns the effective Upstream and Routing state 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) { +) (*config.Config, []string, map[string]adminstate.RoutingState, error) { if supervisor == nil || nilInterface(supervisor.configuration) { - return nil, nil, ErrInvalidSchedulerRunner + return nil, nil, nil, ErrInvalidSchedulerRunner } if supervisor.state == nil { configuration := supervisor.configuration.Current() if configuration == nil { - return nil, nil, ErrInvalidSchedulerRunner + return nil, nil, nil, ErrInvalidSchedulerRunner } - return configuration, enabledUpstreamNames(configuration), nil + return configuration, enabledUpstreamNames(configuration), nil, nil } configurationSource, ok := supervisor.configuration.(SchedulerConfigurationSource) if !ok || nilInterface(configurationSource) || nilInterface(supervisor.state) { - return nil, nil, ErrInvalidSchedulerRunner + return nil, nil, nil, ErrInvalidSchedulerRunner } configuration, revision := configurationSource.Snapshot() if configuration == nil || revision == 0 { - return nil, nil, ErrInvalidSchedulerRunner + return nil, nil, nil, ErrInvalidSchedulerRunner } snapshot, err := supervisor.state.Snapshot(ctx) if err != nil { - return nil, nil, err + return nil, nil, nil, err } if snapshot.Config == nil || snapshot.Config.Revision != revision { - return nil, nil, ErrInvalidSchedulerRunner + return nil, nil, nil, ErrInvalidSchedulerRunner + } + states, err := newSchedulerAdminStates(configuration, snapshot) + if err != nil { + return nil, nil, nil, err + } + names, err := enabledUpstreamNamesForState(configuration, states.upstreams) + if err != nil { + return nil, nil, nil, err + } + return configuration, names, states.routings, nil +} + +type schedulerAdminStates struct { + upstreams map[string]adminstate.UpstreamState + routings map[string]adminstate.RoutingState +} + +func newSchedulerAdminStates( + configuration *config.Config, + snapshot adminstate.Snapshot, +) (schedulerAdminStates, error) { + if configuration == nil { + return schedulerAdminStates{}, ErrInvalidSchedulerRunner + } + states := schedulerAdminStates{ + upstreams: make(map[string]adminstate.UpstreamState, len(snapshot.Upstreams)), + routings: make(map[string]adminstate.RoutingState, len(snapshot.Routings)), } - states := make(map[string]adminstate.UpstreamState, len(snapshot.Upstreams)) for _, upstream := range snapshot.Upstreams { if upstream.Name == "" || upstream.Revision == 0 { - return nil, nil, ErrInvalidSchedulerRunner + return schedulerAdminStates{}, ErrInvalidSchedulerRunner } - if _, duplicate := states[upstream.Name]; duplicate { - return nil, nil, ErrInvalidSchedulerRunner + if _, duplicate := states.upstreams[upstream.Name]; duplicate { + return schedulerAdminStates{}, ErrInvalidSchedulerRunner } - states[upstream.Name] = upstream + states.upstreams[upstream.Name] = upstream } - names, err := enabledUpstreamNamesForState(configuration, states) - if err != nil { - return nil, nil, err + for _, routing := range snapshot.Routings { + if routing.Name == "" || routing.Revision == 0 { + return schedulerAdminStates{}, ErrInvalidSchedulerRunner + } + if _, duplicate := states.routings[routing.Name]; duplicate { + return schedulerAdminStates{}, ErrInvalidSchedulerRunner + } + states.routings[routing.Name] = routing } - return configuration, names, nil + for _, routing := range configuration.Routing { + if _, exists := states.routings[routing.Name]; !exists { + return schedulerAdminStates{}, ErrInvalidSchedulerRunner + } + } + return states, nil } func (supervisor *ConfiguredSchedulerSupervisor) upstreamDueGroups( @@ -396,13 +433,17 @@ type configuredTargetProfile struct { TargetURL string } -func configuredTargetProfiles(configuration *config.Config, upstreamID string) []configuredTargetProfile { +func configuredTargetProfiles( + configuration *config.Config, + upstreamID string, + routings map[string]adminstate.RoutingState, +) []configuredTargetProfile { if configuration == nil || upstreamID == "" { return nil } profiles := make([]configuredTargetProfile, 0) for _, route := range configuration.Routing { - if !route.Enabled || !containsString(route.Upstreams, upstreamID) { + if !route.Enabled || !containsString(route.Upstreams, upstreamID) || (routings != nil && !routings[route.Name].Enabled) { continue } for _, targetURL := range route.Check.Targets { diff --git a/internal/controller/health/scheduler_runner_test.go b/internal/controller/health/scheduler_runner_test.go index c224231..1c9c6a2 100644 --- a/internal/controller/health/scheduler_runner_test.go +++ b/internal/controller/health/scheduler_runner_test.go @@ -191,6 +191,7 @@ func TestConfiguredSchedulerSupervisorSkipsAdminDisabledUpstream(t *testing.T) { {Name: "provider-a", Enabled: false, Revision: 11}, {Name: "provider-b", Enabled: true, Revision: 12}, }, + Routings: []adminstate.RoutingState{{Name: "checkout", Enabled: true, Revision: 13}}, }} sink := &taskSinkStub{} supervisor, err := NewConfiguredSchedulerSupervisor( @@ -209,6 +210,39 @@ func TestConfiguredSchedulerSupervisorSkipsAdminDisabledUpstream(t *testing.T) { } } +func TestConfiguredSchedulerSupervisorSkipsAdminDisabledRoutingTarget(t *testing.T) { + now := time.Date(2026, 8, 7, 10, 15, 0, 0, time.UTC) + configuration := schedulerConfiguration(now, "provider-a") + 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}}, + 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}, + }}, + }, + }} + 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}, + Upstreams: []adminstate.UpstreamState{{Name: "provider-a", Enabled: true, Revision: 11}}, + Routings: []adminstate.RoutingState{{Name: "checkout", Enabled: false, Revision: 12}}, + }}, + ) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + if result, err := supervisor.Tick(context.Background()); err != nil || result != (TickResult{Planned: 1, Offered: 1}) || + source.sources["provider-a"].dueCalls != 1 || source.sources["provider-a"].targetCalls != 0 { + t.Fatalf("Tick() = (%+v, %v); source=%+v", result, err, source.sources["provider-a"]) + } +} + func TestConfiguredSchedulerSupervisorFailsClosedOnAdminRevisionMismatch(t *testing.T) { now := time.Date(2026, 8, 7, 10, 30, 0, 0, time.UTC) configuration := schedulerConfiguration(now, "provider-a") @@ -249,6 +283,30 @@ func TestConfiguredSchedulerSupervisorFailsClosedOnIncompleteAdminState(t *testi } } +func TestConfiguredSchedulerSupervisorFailsClosedOnMissingAdminRoutingState(t *testing.T) { + now := time.Date(2026, 8, 7, 10, 50, 0, 0, time.UTC) + configuration := schedulerConfiguration(now, "provider-a") + configuration.Routing = []config.Routing{{Name: "checkout", Enabled: true, Upstreams: []string{"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}, + Upstreams: []adminstate.UpstreamState{{Name: "provider-a", Enabled: true, Revision: 11}}, + }}, + ) + 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( diff --git a/progress.md b/progress.md index 9f39abd..f1aadba 100644 --- a/progress.md +++ b/progress.md @@ -16,7 +16,7 @@ 原子停用路由并刷新完整 Snapshot。定向测试覆盖无 Provider 读取、末端、回绕和跨副本竞争。 - Health Scheduler 已与 PostgreSQL Admin 管理态对齐：每轮以同一配置 revision 合并启用状态， 被管理态停用的 Upstream 不再读取 Redis due-index，也不会创建 BASIC、EGRESS、TARGET 任务； - revision 不匹配、状态缺失或重复时失败关闭。 + 管理态停用 Routing 则仅阻止该 Routing 的 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 1fb15c6..c9f5bd7 100644 --- a/task_plan.md +++ b/task_plan.md @@ -48,7 +48,8 @@ Prometheus 指标。Admin 成功提交的 Upstream 启停、Routing 切换和配置发布现会经 公用广播器立即刷新本进程所有 Worker 完整 Snapshot；跨 Controller 副本仍以定时刷新 收敛。Health Scheduler 现以相同 revision 合并配置与管理态启用状态，管理态停用会停止 - 该 Upstream 的新 BASIC/EGRESS/TARGET 任务。后续补齐更完整的运行态可观测闭环。 + 该 Upstream 的新 BASIC/EGRESS/TARGET 任务，停用 Routing 仅停止其新 TARGET 任务。后续补齐 + 更完整的运行态可观测闭环。 ## 串并行关系