youfak/proxy-pool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: advance sequential routes past disabled upstreams

Browse Source
This commit is contained in:
youfak 2026-08-07 16:41:00 +08:00
parent 7e73b5ad1c
commit 5e748c6325
8 changed files with 242 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -126,8 +126,8 @@ flowchart LR
Redis 会话栅栏，以及 Gateway Outcome 上报的有界队列、序列确认与重试；
Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5
Checker 探测和
Observation 状态归并；Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤、
末端 `stop` 的 CAS 路由停用和 Snapshot 即时刷新。
Observation 状态归并；Provider 连续空结果与管理态禁用当前项触发的 Sequential CAS
自动推进、禁用候选过滤、末端路由停用和 Snapshot 即时刷新。
- **部分完成**：Kubernetes 运行时 mTLS Overlay；Compose 已具备本地的
Controller/Gateway/Checker mTLS 运行链路。
- **待完成**：故障演练和代表性集群压测；现有 HTTP、CONNECT 长连接和 Extract

5
docs/configuration/reference.md
View File

@ -492,6 +492,11 @@ Upstream 且 `endBehavior: stop` 时，会以当前 Upstream 的 CAS 条件原
Routing。停用后的 Gateway 规则按既有 `onUnavailable.action` 执行 `reject`、`wait`
或 `direct`，直到一次配置重载提交新的管理快照。
管理态禁用当前 Sequential Upstream 不等待新的 Provider 空结果：Controller 会按
配置顺序以相同的 `ExpectedCurrent` CAS 推进到下一个启用 Upstream，并立即刷新完整
Snapshot。若当前项之后没有启用候选，只有 `endBehavior: loop` 会回绕搜索更早的候选；
`stop` 与 `stayLast` 都会原子停用 Routing，因为已禁用的当前项不能继续承接新分配。
## 8. Upstream
```yaml

2
docs/requirements/traceability.md
View File

@ -20,7 +20,7 @@
| ROUTE-001 | Routing 自上而下匹配，首条命中停止 | 3534-3798, 5825-6467 | `rule.go` 与不可变/首命中单测 |
| ROUTE-002 | Routing 与 Upstream 生命周期解耦 | 3534-3798 | 包依赖与配置模型 |
| ROUTE-003 | 支持 sequential、random、roundRobin、weighted、leastConnections | 5825-6467 | 五种领域策略、同版本 Gateway Snapshot 派发和定向测试已完成；Distribution 接线待完成 |
| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；Provider Stats 对连续空结果分配单调代次，Controller `SequentialCoordinator` 在独立有界循环中读取权威配置/管理快照，并通过 `ExpectedCurrent` CAS 自动切换。禁用候选会跳过，重复 Tick、循环后旧代次和并发 Tick 均不会再次切换；末端 `stop` 使用 `DisableRouting` 的同一 CAS 原子停用路由，并与审计/Outbox 同事务提交；两种成功动作都会广播完整 Snapshot |
| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | Provider Stats 对连续空结果分配单调代次，Controller `SequentialCoordinator` 在独立有界循环中读取权威配置/管理快照，并通过 `ExpectedCurrent` CAS 自动切换。禁用候选会跳过；若当前项被管理态禁用，协调器不读取 Provider Stats，按配置顺序推进到下一启用项，只有 `loop` 可回绕，`stop`/`stayLast` 无后继时通过 `DisableRouting` 原子停用路由。重复 Tick、旧代次及跨 Controller 并发 Tick 均不会重复变更；成功切换或停用都会广播完整 Snapshot，并与审计/Outbox 同事务提交。|
| ROUTE-005 | 空计数属于 Upstream，当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 |
| ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽，不因切换直接丢弃 | 6618-6641 | Routing 成功切换后立即发布完整快照；Sequential 仅将新分配切到新的 CurrentUpstream，旧 Proxy 仍保留在快照，既有 Active/Reserved 由本地运行态自然归零。共享 Upstream 不按单 Routing 强制 Drain，避免影响其他 Routing |
| ROUTE-007 | 无可用 Upstream 时显式 reject、wait 或 direct，默认 reject | 5075-5294, 6743-6760 | Gateway 已实现 reject、带 `wait_timeout` 的本地容量等待与经 TargetPolicy 的 direct；Distribution 接线和默认化策略待完成 |

145
internal/controller/routing/sequential.go
View File

@ -26,6 +26,8 @@ const (
autoSwitchActor = "proxy-controller"
autoSwitchReason = "consecutive empty provider fetches reached routing threshold"
autoStopReason = "consecutive empty provider fetches reached terminal sequential end"
disabledSwitchReason = "current sequential upstream is disabled"
disabledStopReason = "current sequential upstream is disabled and no eligible successor remains"
)
// ConfigurationSource supplies one immutable configuration and its matching
@ -165,7 +167,22 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult,
continue
}
state, exists := routings[route.Name]
if !exists || !state.Enabled || !upstreamEnabled(configuration, upstreams, state.CurrentUpstream) {
if !exists || !state.Enabled {
continue
}
if !upstreamEnabled(configuration, upstreams, state.CurrentUpstream) {
transition := disabledCurrentTransition(route, state.CurrentUpstream, upstreams, configuration)
mutated, mutateErr := coordinator.applyTransition(
ctx, route.Name, state.CurrentUpstream, transition, "disabled", state.Revision, disabledSwitchReason, disabledStopReason,
)
if mutateErr != nil {
if errors.Is(mutateErr, adminstate.ErrConflict) {
continue
}
return result, mutateErr
}
result.Switched += mutated.Switched
result.Stopped += mutated.Stopped
continue
}
read := coordinator.stats.ReadProviderStats([]string{state.CurrentUpstream})
@ -183,12 +200,9 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult,
continue
}
transition := nextTransition(route, state.CurrentUpstream, upstreams, configuration)
if transition.stop {
mutation, mutateErr := coordinator.state.DisableRouting(ctx, adminstate.DisableRoutingCommand{
RequestID: requestID("stop", route.Name, state.CurrentUpstream, stats.EmptyGeneration),
Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Reason: autoStopReason,
})
mutated, mutateErr := coordinator.applyTransition(
ctx, route.Name, state.CurrentUpstream, transition, "empty", stats.EmptyGeneration, autoSwitchReason, autoStopReason,
)
if mutateErr != nil {
if errors.Is(mutateErr, adminstate.ErrConflict) {
coordinator.processed[key] = stats.EmptyGeneration
@ -197,41 +211,62 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult,
return result, mutateErr
}
coordinator.processed[key] = stats.EmptyGeneration
if mutation.Changed {
result.Stopped++
if coordinator.refresh != nil {
coordinator.refresh.NotifySnapshotRefresh()
}
}
continue
}
if transition.target == "" {
coordinator.processed[key] = stats.EmptyGeneration
continue
}
mutation, mutateErr := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{
RequestID: requestID("switch", route.Name, state.CurrentUpstream, stats.EmptyGeneration),
Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
Name: route.Name, ExpectedCurrent: state.CurrentUpstream, Target: transition.target, Reason: autoSwitchReason,
})
if mutateErr != nil {
if errors.Is(mutateErr, adminstate.ErrConflict) {
coordinator.processed[key] = stats.EmptyGeneration
continue
}
return result, mutateErr
}
coordinator.processed[key] = stats.EmptyGeneration
if mutation.Changed {
result.Switched++
if coordinator.refresh != nil {
coordinator.refresh.NotifySnapshotRefresh()
}
}
result.Switched += mutated.Switched
result.Stopped += mutated.Stopped
}
return result, nil
}
// applyTransition centralizes the mutation/refresh side effect for both the
// Provider-empty and management-disabled paths. ExpectedCurrent remains the
// only mutation fence, so concurrent Controller replicas cannot skip states.
func (coordinator *SequentialCoordinator) applyTransition(
ctx context.Context,
routeName string,
current string,
transition sequentialTransition,
origin string,
version uint64,
switchReason string,
stopReason string,
) (TickResult, error) {
if transition.stop {
mutation, err := coordinator.state.DisableRouting(ctx, adminstate.DisableRoutingCommand{
RequestID: requestID(origin+"-stop", routeName, current, version),
Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
Name: routeName, ExpectedCurrent: current, Reason: stopReason,
})
if err != nil {
return TickResult{}, err
}
if !mutation.Changed {
return TickResult{}, nil
}
if coordinator.refresh != nil {
coordinator.refresh.NotifySnapshotRefresh()
}
return TickResult{Stopped: 1}, nil
}
if transition.target == "" {
return TickResult{}, nil
}
mutation, err := coordinator.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{
RequestID: requestID(origin+"-switch", routeName, current, version),
Actor: adminstate.Actor{ID: autoSwitchActor}, OccurredAt: coordinator.now().UTC(),
Name: routeName, ExpectedCurrent: current, Target: transition.target, Reason: switchReason,
})
if err != nil {
return TickResult{}, err
}
if !mutation.Changed {
return TickResult{}, nil
}
if coordinator.refresh != nil {
coordinator.refresh.NotifySnapshotRefresh()
}
return TickResult{Switched: 1}, nil
}
type sequentialTransition struct {
target string
stop bool
@ -275,6 +310,40 @@ func nextTransition(
return sequentialTransition{}
}
// disabledCurrentTransition treats an administratively disabled current
// upstream as a permanent advance. stayLast cannot retain a disabled current,
// so only loop may wrap; every other terminal condition stops the Routing.
func disabledCurrentTransition(
route config.Routing,
current string,
states map[string]adminstate.UpstreamState,
configuration *config.Config,
) sequentialTransition {
currentIndex := -1
for index, upstream := range route.Upstreams {
if upstream == current {
currentIndex = index
break
}
}
if currentIndex < 0 {
return sequentialTransition{stop: true}
}
for index := currentIndex + 1; index < len(route.Upstreams); index++ {
if upstreamEnabled(configuration, states, route.Upstreams[index]) {
return sequentialTransition{target: route.Upstreams[index]}
}
}
if route.Strategy.EndBehavior == "loop" {
for index := 0; index < currentIndex; index++ {
if upstreamEnabled(configuration, states, route.Upstreams[index]) {
return sequentialTransition{target: route.Upstreams[index]}
}
}
}
return sequentialTransition{stop: true}
}
func upstreamStates(values []adminstate.UpstreamState) map[string]adminstate.UpstreamState {
result := make(map[string]adminstate.UpstreamState, len(values))
for _, value := range values {

74
internal/controller/routing/sequential_test.go
View File

@ -59,6 +59,40 @@ func TestSequentialCoordinatorStopsWhenDisabledCandidatesLeaveNoAlternative(t *t
}
}
func TestSequentialCoordinatorAdvancesWhenCurrentUpstreamIsDisabledWithoutProviderStats(t *testing.T) {
state := newRoutingState("provider-a", map[string]bool{"provider-a": false, "provider-b": true, "provider-c": true})
stats := &routingStats{byUpstream: map[string]provider.Stats{}}
refresh := &refreshRecorder{}
coordinator := newCoordinator(t, routingConfiguration("stop", []string{"provider-a", "provider-b", "provider-c"}), state, stats, refresh)
if result, err := coordinator.Tick(context.Background()); err != nil || result != (TickResult{Switched: 1}) || state.current() != "provider-b" {
t.Fatalf("Tick(disabled current) = (%+v, %v), current=%q", result, err, state.current())
}
if stats.readCount() != 0 || state.switches != 1 || refresh.count != 1 {
t.Fatalf("provider reads=%d switches=%d refresh=%d, want 0/1/1", stats.readCount(), state.switches, refresh.count)
}
}
func TestSequentialCoordinatorStopsWhenDisabledCurrentHasNoSuccessor(t *testing.T) {
state := newRoutingState("provider-b", map[string]bool{"provider-a": true, "provider-b": false})
coordinator := newCoordinator(t, routingConfiguration("stayLast", []string{"provider-a", "provider-b"}), state,
&routingStats{byUpstream: map[string]provider.Stats{}}, &refreshRecorder{})
if result, err := coordinator.Tick(context.Background()); err != nil || result != (TickResult{Stopped: 1}) || state.enabled() {
t.Fatalf("Tick(disabled terminal current) = (%+v, %v), enabled=%v", result, err, state.enabled())
}
}
func TestSequentialCoordinatorLoopsWhenDisabledCurrentHasEarlierSuccessor(t *testing.T) {
state := newRoutingState("provider-c", map[string]bool{"provider-a": true, "provider-b": false, "provider-c": false})
coordinator := newCoordinator(t, routingConfiguration("loop", []string{"provider-a", "provider-b", "provider-c"}), state,
&routingStats{byUpstream: map[string]provider.Stats{}}, &refreshRecorder{})
if result, err := coordinator.Tick(context.Background()); err != nil || result != (TickResult{Switched: 1}) || state.current() != "provider-a" {
t.Fatalf("Tick(disabled loop current) = (%+v, %v), current=%q", result, err, state.current())
}
}
func TestSequentialCoordinatorConsumesStatsRecorderNotifications(t *testing.T) {
state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true})
stats, err := provider.NewStatsRecorder(2)
@ -153,6 +187,33 @@ func TestSequentialCoordinatorUsesCompareAndSwapUnderConcurrentTicks(t *testing.
}
}
func TestSequentialCoordinatorUsesCompareAndSwapForDisabledCurrentAcrossReplicas(t *testing.T) {
state := newRoutingState("provider-a", map[string]bool{"provider-a": false, "provider-b": true})
configuration := routingConfiguration("stop", []string{"provider-a", "provider-b"})
stats := &routingStats{byUpstream: map[string]provider.Stats{}}
first := newCoordinator(t, configuration, state, stats, &refreshRecorder{})
second := newCoordinator(t, configuration, state, stats, &refreshRecorder{})
var wait sync.WaitGroup
for index := range 32 {
wait.Add(1)
go func() {
defer wait.Done()
coordinator := first
if index%2 == 1 {
coordinator = second
}
if _, err := coordinator.Tick(context.Background()); err != nil {
t.Errorf("Tick() = %v", err)
}
}()
}
wait.Wait()
if state.switches != 1 || state.current() != "provider-b" || !state.enabled() {
t.Fatalf("switches=%d current=%q enabled=%v, want one switch to provider-b", state.switches, state.current(), state.enabled())
}
}
func TestSequentialCoordinatorFailsClosedWhenConfigurationRevisionIsStale(t *testing.T) {
state := newRoutingState("provider-a", map[string]bool{"provider-a": true, "provider-b": true})
state.snapshot.Config.Revision = 8
@ -251,7 +312,10 @@ func newRoutingState(current string, enabled map[string]bool) *routingState {
func (state *routingState) Snapshot(context.Context) (adminstate.Snapshot, error) {
state.mu.Lock()
defer state.mu.Unlock()
return state.snapshot, nil
snapshot := state.snapshot
snapshot.Upstreams = append([]adminstate.UpstreamState(nil), state.snapshot.Upstreams...)
snapshot.Routings = append([]adminstate.RoutingState(nil), state.snapshot.Routings...)
return snapshot, nil
}
func (state *routingState) SwitchRouting(_ context.Context, command adminstate.SwitchRoutingCommand) (adminstate.MutationResult, error) {
@ -296,6 +360,7 @@ func (state *routingState) enabled() bool {
type routingStats struct {
mu sync.Mutex
byUpstream map[string]provider.Stats
reads int
}
type emptyStats struct{}
@ -305,6 +370,7 @@ func (emptyStats) ReadProviderStats([]string) []provider.Stats { return nil }
func (stats *routingStats) ReadProviderStats(upstreams []string) []provider.Stats {
stats.mu.Lock()
defer stats.mu.Unlock()
stats.reads++
result := make([]provider.Stats, len(upstreams))
for index, upstream := range upstreams {
result[index] = stats.byUpstream[upstream]
@ -313,6 +379,12 @@ func (stats *routingStats) ReadProviderStats(upstreams []string) []provider.Stat
return result
}
func (stats *routingStats) readCount() int {
stats.mu.Lock()
defer stats.mu.Unlock()
return stats.reads
}
func (stats *routingStats) set(upstream string, value provider.Stats) {
stats.mu.Lock()
defer stats.mu.Unlock()

28
internal/controller/worker/routing_snapshot_source_test.go
View File

@ -114,6 +114,34 @@ func TestGatewayRoutingSourceDisablesSequentialRuleWhenCurrentUpstreamIsUnavaila
}
}
func TestGatewayRoutingSourcePublishesAdvancedSequentialCurrent(t *testing.T) {
configuration := &config.Config{Routing: []config.Routing{{
Name: "gateway", Enabled: true, Purpose: "gateway", Upstreams: []string{"provider-a", "provider-b"},
Strategy: config.Strategy{Type: "sequential"}, OnUnavailable: config.OnUnavailable{Action: "reject"},
}}, Upstreams: map[string]config.Upstream{
"provider-a": {Enabled: true}, "provider-b": {Enabled: true},
}}
source, err := NewGatewayRoutingSource(
staticGatewayRoutingConfiguration{configuration: configuration, revision: 5},
staticGatewayRoutingState{snapshot: adminstate.Snapshot{
Config: &adminstate.ConfigRevision{Revision: 5},
Upstreams: []adminstate.UpstreamState{{Name: "provider-a", Enabled: false}, {Name: "provider-b", Enabled: true}},
Routings: []adminstate.RoutingState{{Name: "gateway", Enabled: true, CurrentUpstream: "provider-b"}},
}},
)
if err != nil {
t.Fatalf("NewGatewayRoutingSource(): %v", err)
}
rules, err := source.Read(context.Background())
if err != nil {
t.Fatalf("Read(): %v", err)
}
if len(rules) != 1 || !rules[0].GetEnabled() || rules[0].GetStrategy().GetCurrentUpstream() != "provider-b" ||
!reflect.DeepEqual(rules[0].GetUpstreams(), []string{"provider-b"}) {
t.Fatalf("rules = %+v, want enabled Sequential rule on provider-b", rules)
}
}
func TestGatewayRoutingSourceBuildsStaticDirectRuleWithoutUpstreams(t *testing.T) {
configuration := &config.Config{Routing: []config.Routing{{
Name: "direct-api", Enabled: true, Purpose: "gateway", Action: "direct",

19
progress.md
View File

@ -1,5 +1,24 @@
# 项目进度
## 2026-08-07
- Gateway 静态 `routing.action: direct` 已贯通严格配置、控制面协议、完整 Snapshot、
HTTP/CONNECT 转发和目标地址策略。直连请求不会申请 Proxy 容量、创建粘性绑定或上报
Proxy Outcome；`onUnavailable.action: direct` 仍保留为代理无候选时的独立回退语义。
- Loadgen 固定速率模式已修复截止时排队令牌的记账：`generated = requests + dropped`，
防止报告将未执行的请求误记为已发送。
- Compose 已启用 Controller、两个固定身份 Gateway 和 Checker 的本地 mTLS 控制面；
PowerShell/.NET 证书生成器输出 7 天 Controller DNS SAN 与角色隔离的 SPIFFE URI，
私钥目录被 Git 忽略。Checker 新增 `/livez`、`/readyz`、`/metrics`，首次成功领取任务
批次后才 Ready，后续领取失败会撤销 Ready。
- 管理态禁用当前 Sequential Upstream 现不依赖 Provider 空结果：Controller 以
`ExpectedCurrent` CAS 推进到后续启用项，`loop` 可回绕，`stop`/`stayLast` 无后继时
原子停用路由并刷新完整 Snapshot。定向测试覆盖无 Provider 读取、末端、回绕和跨副本竞争。
- 全仓 `go test -count=1 -timeout 60s ./...`、`go vet ./...`、`go build ./...`、
Protobuf descriptor、Kustomize Base 渲染及开发证书 SAN/SPIFFE 校验均通过。Compose
容器端到端启动在拉取 Dockerfile 前端与监控镜像时受 Docker Desktop HTTPS 代理缺失阻断，
未把该环境问题记为运行验证通过。
## 2026-08-02
- 容量可观测性已接入 Provider 补池对账循环：每个成功 Redis 库存读数更新 Controller

5
task_plan.md
View File

@ -72,5 +72,6 @@
Provider 分布式协调和 Worker 运行态 Redis 原语已完成，但 WorkerControlPlane
接收端、Provider Fleet、Gateway、Checker、Loadgen、业务指标、Redis 故障
转移验证与代表性集群压测属于后续实施范围。
- `implementation-plan.md` 当前按 74 个验收项统计；已校正为 59 项完成，
验收项完成率约 79.7%，不等同于生产就绪度。
- `implementation-plan.md` 当前按 74 个验收项统计；验收勾选数不等同于生产就绪度。
Compose 本地 mTLS 控制面运行链已实现并通过配置、证书、单元与构建验证；受本机 Docker
Desktop 镜像 HTTPS 代理缺失影响，容器端到端启动仍待具备镜像网络的环境复核。