From 5e748c632503660383ca08ba86752d73b28f1eb4 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 7 Aug 2026 16:41:00 +0800 Subject: [PATCH] feat: advance sequential routes past disabled upstreams --- README.md | 4 +- docs/configuration/reference.md | 5 + docs/requirements/traceability.md | 2 +- internal/controller/routing/sequential.go | 153 +++++++++++++----- .../controller/routing/sequential_test.go | 74 ++++++++- .../worker/routing_snapshot_source_test.go | 28 ++++ progress.md | 19 +++ task_plan.md | 5 +- 8 files changed, 242 insertions(+), 48 deletions(-) diff --git a/README.md b/README.md index c36d3a2..68e7ebf 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 8d28a66..ae04d7b 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -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 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index b806593..ce9e178 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -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 接线和默认化策略待完成 | diff --git a/internal/controller/routing/sequential.go b/internal/controller/routing/sequential.go index 0ce198d..1dd7bf3 100644 --- a/internal/controller/routing/sequential.go +++ b/internal/controller/routing/sequential.go @@ -22,10 +22,12 @@ import ( var ErrInvalidSequentialCoordinator = errors.New("invalid sequential routing coordinator") const ( - defaultPollInterval = time.Second - autoSwitchActor = "proxy-controller" - autoSwitchReason = "consecutive empty provider fetches reached routing threshold" - autoStopReason = "consecutive empty provider fetches reached terminal sequential end" + defaultPollInterval = time.Second + 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,37 +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, - }) - 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.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, - }) + 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 @@ -222,16 +211,62 @@ func (coordinator *SequentialCoordinator) Tick(ctx context.Context) (TickResult, 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 { diff --git a/internal/controller/routing/sequential_test.go b/internal/controller/routing/sequential_test.go index 7c69c6e..db1b2f4 100644 --- a/internal/controller/routing/sequential_test.go +++ b/internal/controller/routing/sequential_test.go @@ -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() diff --git a/internal/controller/worker/routing_snapshot_source_test.go b/internal/controller/worker/routing_snapshot_source_test.go index 4896bf0..d402109 100644 --- a/internal/controller/worker/routing_snapshot_source_test.go +++ b/internal/controller/worker/routing_snapshot_source_test.go @@ -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", diff --git a/progress.md b/progress.md index 5124c80..ed85ee0 100644 --- a/progress.md +++ b/progress.md @@ -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 diff --git a/task_plan.md b/task_plan.md index 7902502..706e2c7 100644 --- a/task_plan.md +++ b/task_plan.md @@ -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 代理缺失影响，容器端到端启动仍待具备镜像网络的环境复核。