feat: honor admin routing state in target scheduling
This commit is contained in:
youfak
parent
351836b5e4
commit
a191ca5837
@ -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 事实
|
||||
|
||||
@ -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`
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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 代理缺失阻断，
|
||||
|
||||
@ -48,7 +48,8 @@
|
||||
Prometheus 指标。Admin 成功提交的 Upstream 启停、Routing 切换和配置发布现会经
|
||||
公用广播器立即刷新本进程所有 Worker 完整 Snapshot；跨 Controller 副本仍以定时刷新
|
||||
收敛。Health Scheduler 现以相同 revision 合并配置与管理态启用状态，管理态停用会停止
|
||||
该 Upstream 的新 BASIC/EGRESS/TARGET 任务。后续补齐更完整的运行态可观测闭环。
|
||||
该 Upstream 的新 BASIC/EGRESS/TARGET 任务，停用 Routing 仅停止其新 TARGET 任务。后续补齐
|
||||
更完整的运行态可观测闭环。
|
||||
|
||||
## 串并行关系
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user