diff --git a/README.md b/README.md index e4401c7..f4cdf22 100644 --- a/README.md +++ b/README.md @@ -45,8 +45,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行 有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 及 Gateway Routing 快照。Gateway 会将 Routing 与 Proxy 原子编译为同一内存 View，并只按当前未过期 - View 匹配请求；`proxy-gateway` 命令、凭据分发，以及策略选择与 `onUnavailable` 派发语义 - 仍待装配。 + View 匹配请求，并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least + Connections 选择上游；`proxy-gateway` 命令、凭据分发和 `onUnavailable` 派发语义仍待装配。 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 @@ -85,9 +85,8 @@ flowchart LR Redis 会话栅栏。 - **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership 与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。 -- **待完成**：带凭据 Proxy 分发、Gateway 进程装配、Routing 策略选择与 - `onUnavailable`、Outcome 上报、Checker 调度与健康状态链，以及 loadgen 和代表性 - 集群压测。 +- **待完成**：带凭据 Proxy 分发、Gateway 进程装配、`onUnavailable`、Outcome 上报、 + Checker 调度与健康状态链，以及 loadgen 和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index af4415a..2fda535 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -31,8 +31,9 @@ Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有 `usable_until`；带凭据 Proxy 会 fail-closed，直到凭据材料分发完成。完整 Snapshot 已从配置原始顺序和 Admin 当前状态合成 Gateway Routing，并与 Proxy 一起纳入 checksum； Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本地 View，动态 Router 只匹配 -当前未过期 View。策略选择、`onUnavailable`、增量、Gateway 进程装配、Outcome 与 Checker -闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；`100,000 QPS` 仍是未验证的设计目标。 +当前未过期 View；派发器已按五种策略从该 View 选择上游，且在 Proxy 容量耗尽时只在该 +View 的其余候选中回退。`onUnavailable`、增量、Gateway 进程装配、Outcome 与 Checker 闭环 +尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；`100,000 QPS` 仍是未验证的设计目标。 `WatchSnapshots` 建立时校验当前 session；每次签发快照引用时也把 `session_id` 交给 Redis 原子校验。重复 Register 会同时清除旧 Runtime 和已签发引用，因此迟到的 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index f3d7b01..ca422b2 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -220,6 +220,7 @@ Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已 提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Gateway Routing payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 checksum；Gateway 已将其与 Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。策略选择、`onUnavailable`、 +派发器的五种上游选择已接入该 View，并在容量耗尽时在同版本候选中回退；`onUnavailable`、 凭据分发、Gateway 命令与 Outcome 上报仍未实现。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 6f4fee8..bded8fe 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -100,13 +100,13 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt SPIFFE mTLS 校验和 Controller 生命周期接线已完成；Redis Provider Leader、 分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。 6. Worker 基础网络快照流、无凭据引用 Proxy/Gateway Routing payload、Gateway Snapshot - 客户端及同版本 Routing 编译/动态匹配已完成；策略选择、`onUnavailable`、凭据分发与 - Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 + 客户端、同版本 Routing 编译/动态匹配和五种策略上游选择已完成；`onUnavailable`、凭据 + 分发与 Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 7. Checker 调度、探测器和健康 reducer。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 -10. 将五种 Routing 策略和 `onUnavailable` 接入 Gateway/Distribution 运行链， - 补齐 Sequential 持久化恢复、跨实例 CAS 和 disabled candidate 语义。 +10. 将 `onUnavailable` 接入 Gateway/Distribution 运行链，补齐 Sequential 持久化恢复、 + 跨实例 CAS 和 disabled candidate 语义。 11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测；短 TTL Proxy 的零计数运行态已在后续 Snapshot Apply 中回收。 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 34a13cc..355def3 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -19,7 +19,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/Distribution 运行链的接线待完成 | +| ROUTE-003 | 支持 sequential、random、roundRobin、weighted、leastConnections | 5825-6467 | 五种领域策略、同版本 Gateway Snapshot 派发和定向测试已完成；Distribution 接线待完成 | | ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；持久化恢复和跨实例 CAS 待完成 | | ROUTE-005 | 空计数属于 Upstream，当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 | | ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽，不因切换直接丢弃 | 6618-6641 | 通用 ownership Drain/ACK 原语已测试；Routing 切换到 Drain 的编排待完成 | diff --git a/internal/domain/routing/strategy.go b/internal/domain/routing/strategy.go index a2cd2ac..11483c7 100644 --- a/internal/domain/routing/strategy.go +++ b/internal/domain/routing/strategy.go @@ -4,6 +4,7 @@ import ( "errors" "math/rand" "sync" + "sync/atomic" ) var ErrNoCandidate = errors.New("no eligible routing candidate") @@ -24,7 +25,7 @@ type RandomSource interface { } type randomSelector struct { - source *synchronizedRandomSource + source RandomSource } type globalRandomSource struct{} @@ -73,8 +74,7 @@ func (s *randomSelector) Select(candidates []Candidate) (Candidate, error) { } type roundRobinSelector struct { - mu sync.Mutex - next int + next atomic.Uint64 } func NewRoundRobin() Selector { @@ -86,22 +86,36 @@ func (s *roundRobinSelector) Select(candidates []Candidate) (Candidate, error) { return Candidate{}, ErrNoCandidate } - s.mu.Lock() - defer s.mu.Unlock() - start := s.next % len(candidates) - for offset := range len(candidates) { - selected := (start + offset) % len(candidates) - if !candidates[selected].Eligible { - continue + for { + next := s.next.Load() + start := int(next % uint64(len(candidates))) + for offset := range len(candidates) { + selected := (start + offset) % len(candidates) + if !candidates[selected].Eligible { + continue + } + if s.next.CompareAndSwap(next, uint64(selected+1)) { + return candidates[selected], nil + } + break + } + if allCandidatesIneligible(candidates) { + return Candidate{}, ErrNoCandidate } - s.next = (selected + 1) % len(candidates) - return candidates[selected], nil } - return Candidate{}, ErrNoCandidate +} + +func allCandidatesIneligible(candidates []Candidate) bool { + for _, candidate := range candidates { + if candidate.Eligible { + return false + } + } + return true } type weightedSelector struct { - source *synchronizedRandomSource + source RandomSource } func NewWeighted(source ...RandomSource) Selector { @@ -159,10 +173,9 @@ func (leastConnectionsSelector) Select(candidates []Candidate) (Candidate, error return selected, nil } -func newSynchronizedRandomSource(sources []RandomSource) *synchronizedRandomSource { - var source RandomSource = globalRandomSource{} - if len(sources) > 0 && sources[0] != nil { - source = sources[0] +func newSynchronizedRandomSource(sources []RandomSource) RandomSource { + if len(sources) == 0 || sources[0] == nil { + return globalRandomSource{} } - return &synchronizedRandomSource{source: source} + return &synchronizedRandomSource{source: sources[0]} } diff --git a/internal/gateway/dispatch/dispatcher.go b/internal/gateway/dispatch/dispatcher.go index a109587..ed99295 100644 --- a/internal/gateway/dispatch/dispatcher.go +++ b/internal/gateway/dispatch/dispatcher.go @@ -2,6 +2,7 @@ package dispatch import ( "errors" + "sync" "sync/atomic" "time" @@ -36,12 +37,24 @@ func (l *Lease) Cancel() error { return l.reserved.Cancel() } func (l *Lease) Release() error { return l.reserved.Release() } type Dispatcher struct { - store *snapshot.Store - cursor atomic.Uint64 + store *snapshot.Store + cursor atomic.Uint64 + random routing.RandomSource + routingState atomic.Pointer[routingSelectorState] } -func New(store *snapshot.Store) *Dispatcher { - return &Dispatcher{store: store} +type routingSelectorState struct { + epoch uint64 + version uint64 + selectors sync.Map +} + +func New(store *snapshot.Store, randomSources ...routing.RandomSource) *Dispatcher { + dispatcher := &Dispatcher{store: store} + if len(randomSources) > 0 { + dispatcher.random = randomSources[0] + } + return dispatcher } func (d *Dispatcher) Acquire(request Request) (*Lease, error) { @@ -55,11 +68,121 @@ func (d *Dispatcher) Acquire(request Request) (*Lease, error) { if request.Now.IsZero() { request.Now = time.Now().UTC() } + if request.Strategy.Type != "" { + return d.acquireRouted(view, request) + } + return d.acquireFromUpstreams(view, request, request.Upstreams) +} + +func (d *Dispatcher) acquireRouted(view *snapshot.View, request Request) (*Lease, error) { + if request.Strategy.Type == routing.StrategySequential { + if request.Strategy.CurrentUpstream == "" { + return nil, ErrNoCandidate + } + return d.acquireFromUpstreams(view, request, []string{request.Strategy.CurrentUpstream}) + } + candidates, err := d.routingCandidates(view, request) + if err != nil { + return nil, ErrNoCandidate + } + selector, err := d.routingSelector(view, request) + if err != nil { + return nil, ErrNoCandidate + } + for range candidates { + candidate, selectErr := selector.Select(candidates) + if selectErr != nil { + return nil, ErrNoCandidate + } + lease, acquireErr := d.acquireFromUpstreams(view, request, []string{candidate.Name}) + if acquireErr == nil { + return lease, nil + } + for index := range candidates { + if candidates[index].Name == candidate.Name { + candidates[index].Eligible = false + break + } + } + } + return nil, ErrNoCandidate +} + +func (d *Dispatcher) routingCandidates(view *snapshot.View, request Request) ([]routing.Candidate, error) { + if len(request.Upstreams) == 0 { + return nil, ErrNoCandidate + } + result := make([]routing.Candidate, 0, len(request.Upstreams)) + seen := make(map[string]struct{}, len(request.Upstreams)) + for _, upstream := range request.Upstreams { + if upstream == "" { + return nil, ErrNoCandidate + } + if _, duplicate := seen[upstream]; duplicate { + return nil, ErrNoCandidate + } + seen[upstream] = struct{}{} + candidate := routing.Candidate{Name: upstream, Eligible: true} + switch request.Strategy.Type { + case routing.StrategyWeighted: + weight, exists := request.Strategy.Weights[upstream] + if !exists || weight == 0 || uint64(weight) > uint64(maxInt()) { + return nil, ErrNoCandidate + } + candidate.Weight = int(weight) + case routing.StrategyLeastConnections: + candidate.Active, candidate.Eligible = view.UpstreamLoad(snapshot.Query{ + Now: request.Now, Scheme: request.Scheme, RequiredTags: request.RequiredTags, + Exclude: request.Exclude, SafetyMargin: request.SafetyMargin, + }, upstream) + } + result = append(result, candidate) + } + return result, nil +} + +func (d *Dispatcher) routingSelector(view *snapshot.View, request Request) (routing.Selector, error) { + state := d.selectorState(view) + key := request.RoutingName + "\x00" + string(request.Strategy.Type) + if existing, found := state.selectors.Load(key); found { + return existing.(routing.Selector), nil + } + var selector routing.Selector + switch request.Strategy.Type { + case routing.StrategyRandom: + selector = routing.NewRandom(d.random) + case routing.StrategyRoundRobin: + selector = routing.NewRoundRobin() + case routing.StrategyWeighted: + selector = routing.NewWeighted(d.random) + case routing.StrategyLeastConnections: + selector = routing.NewLeastConnections() + default: + return nil, ErrNoCandidate + } + actual, _ := state.selectors.LoadOrStore(key, selector) + return actual.(routing.Selector), nil +} + +func (d *Dispatcher) selectorState(view *snapshot.View) *routingSelectorState { + for { + current := d.routingState.Load() + if current != nil && current.epoch == view.Epoch && current.version == view.Version { + return current + } + next := &routingSelectorState{epoch: view.Epoch, version: view.Version} + if d.routingState.CompareAndSwap(current, next) { + return next + } + } +} + +func (d *Dispatcher) acquireFromUpstreams(view *snapshot.View, request Request, upstreams []string) (*Lease, error) { selection := view.Select(snapshot.Query{ Now: request.Now, Scheme: request.Scheme, - Upstreams: request.Upstreams, + Upstreams: upstreams, RequiredTags: request.RequiredTags, Exclude: request.Exclude, SafetyMargin: request.SafetyMargin, @@ -87,3 +210,7 @@ func (d *Dispatcher) Acquire(request Request) (*Lease, error) { } return nil, ErrNoCandidate } + +func maxInt() int { + return int(^uint(0) >> 1) +} diff --git a/internal/gateway/dispatch/dispatcher_test.go b/internal/gateway/dispatch/dispatcher_test.go index 61bbd62..89070c6 100644 --- a/internal/gateway/dispatch/dispatcher_test.go +++ b/internal/gateway/dispatch/dispatcher_test.go @@ -9,6 +9,7 @@ import ( "time" proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/snapshot" ) @@ -112,6 +113,185 @@ func TestAcquireNeverOversubscribesSnapshotProxy(t *testing.T) { } } +func TestAcquireUsesSequentialCurrentUpstream(t *testing.T) { + store := routingStrategyStore(t) + dispatcher := New(store) + lease, err := dispatcher.Acquire(Request{ + Now: time.Now(), RoutingName: "sequential", Upstreams: []string{"upstream-a", "upstream-b"}, + Strategy: routing.Strategy{Type: routing.StrategySequential, CurrentUpstream: "upstream-b"}, + }) + if err != nil { + t.Fatalf("Acquire(): %v", err) + } + if lease.Proxy.SourceUpstream != "upstream-b" { + t.Fatalf("selected upstream = %q, want upstream-b", lease.Proxy.SourceUpstream) + } + if err := lease.Cancel(); err != nil { + t.Fatalf("Cancel(): %v", err) + } +} + +func TestAcquireRoundRobinCyclesRoutingUpstreams(t *testing.T) { + store := routingStrategyStore(t) + dispatcher := New(store) + request := Request{ + Now: time.Now(), RoutingName: "round-robin", Upstreams: []string{"upstream-a", "upstream-b"}, + Strategy: routing.Strategy{Type: routing.StrategyRoundRobin}, + } + for index, want := range []string{"upstream-a", "upstream-b", "upstream-a"} { + lease, err := dispatcher.Acquire(request) + if err != nil { + t.Fatalf("Acquire(%d): %v", index, err) + } + if lease.Proxy.SourceUpstream != want { + t.Fatalf("Acquire(%d) upstream = %q, want %q", index, lease.Proxy.SourceUpstream, want) + } + if err := lease.Cancel(); err != nil { + t.Fatalf("Cancel(%d): %v", index, err) + } + } +} + +func TestAcquireResetsRoundRobinStateWhenSnapshotVersionChanges(t *testing.T) { + store := routingStrategyStore(t) + dispatcher := New(store) + request := Request{ + Now: time.Now(), RoutingName: "round-robin", Upstreams: []string{"upstream-a", "upstream-b"}, + Strategy: routing.Strategy{Type: routing.StrategyRoundRobin}, + } + first, err := dispatcher.Acquire(request) + if err != nil { + t.Fatalf("Acquire(first): %v", err) + } + if first.Proxy.SourceUpstream != "upstream-a" { + t.Fatalf("first upstream = %q, want upstream-a", first.Proxy.SourceUpstream) + } + if err := first.Cancel(); err != nil { + t.Fatalf("Cancel(first): %v", err) + } + + next := snapshot.Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 2, Full: true, + Proxies: []proxyDomain.Proxy{ + {ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, SourceUpstream: "upstream-a", State: proxyDomain.StateAvailable, MaxConcurrency: 1}, + {ID: "proxy-b", Scheme: proxyDomain.SchemeHTTP, SourceUpstream: "upstream-b", State: proxyDomain.StateAvailable, MaxConcurrency: 1}, + }, + } + next.Checksum = snapshot.Checksum(next.Proxies) + if err := store.Apply(next); err != nil { + t.Fatalf("Apply(next): %v", err) + } + second, err := dispatcher.Acquire(request) + if err != nil { + t.Fatalf("Acquire(second): %v", err) + } + if second.Proxy.SourceUpstream != "upstream-a" { + t.Fatalf("second upstream = %q, want reset upstream-a", second.Proxy.SourceUpstream) + } + if err := second.Cancel(); err != nil { + t.Fatalf("Cancel(second): %v", err) + } +} + +func TestAcquireFallsBackWhenSelectedRoutingUpstreamIsFull(t *testing.T) { + store := routingStrategyStore(t) + var held *proxyDomain.Reservation + for _, entry := range store.Current().Entries { + if entry.Proxy.SourceUpstream != "upstream-a" { + continue + } + reservation, ok := entry.Runtime.Reserve() + if !ok { + t.Fatal("Reserve(upstream-a) = false, want true") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit(upstream-a): %v", err) + } + held = reservation + break + } + if held == nil { + t.Fatal("upstream-a reservation was not created") + } + defer func() { + if err := held.Release(); err != nil { + t.Errorf("Release(upstream-a): %v", err) + } + }() + + lease, err := New(store).Acquire(Request{ + Now: time.Now(), RoutingName: "round-robin", Upstreams: []string{"upstream-a", "upstream-b"}, + Strategy: routing.Strategy{Type: routing.StrategyRoundRobin}, + }) + if err != nil { + t.Fatalf("Acquire(): %v", err) + } + if lease.Proxy.SourceUpstream != "upstream-b" { + t.Fatalf("selected upstream = %q, want upstream-b", lease.Proxy.SourceUpstream) + } + if err := lease.Cancel(); err != nil { + t.Fatalf("Cancel(): %v", err) + } +} + +func TestAcquireUsesWeightedRoutingStrategy(t *testing.T) { + store := routingStrategyStore(t) + dispatcher := New(store, &fixedRoutingRandom{values: []int{0, 2, 3}}) + request := Request{ + Now: time.Now(), RoutingName: "weighted", Upstreams: []string{"upstream-a", "upstream-b"}, + Strategy: routing.Strategy{Type: routing.StrategyWeighted, Weights: map[string]uint32{"upstream-a": 3, "upstream-b": 1}}, + } + for index, want := range []string{"upstream-a", "upstream-a", "upstream-b"} { + lease, err := dispatcher.Acquire(request) + if err != nil { + t.Fatalf("Acquire(%d): %v", index, err) + } + if lease.Proxy.SourceUpstream != want { + t.Fatalf("Acquire(%d) upstream = %q, want %q", index, lease.Proxy.SourceUpstream, want) + } + if err := lease.Cancel(); err != nil { + t.Fatalf("Cancel(%d): %v", index, err) + } + } +} + +func TestAcquireUsesLeastConnectionsAndSkipsFullUpstream(t *testing.T) { + store := routingStrategyStore(t) + view := store.Current() + for _, entry := range view.Entries { + if entry.Proxy.SourceUpstream != "upstream-a" { + continue + } + reservation, ok := entry.Runtime.Reserve() + if !ok { + t.Fatal("Reserve(upstream-a) = false, want true") + } + if err := reservation.Commit(); err != nil { + t.Fatalf("Commit(upstream-a): %v", err) + } + defer func() { + if err := reservation.Release(); err != nil { + t.Errorf("Release(upstream-a): %v", err) + } + }() + break + } + + lease, err := New(store).Acquire(Request{ + Now: time.Now(), RoutingName: "least", Upstreams: []string{"upstream-a", "upstream-b"}, + Strategy: routing.Strategy{Type: routing.StrategyLeastConnections}, + }) + if err != nil { + t.Fatalf("Acquire(): %v", err) + } + if lease.Proxy.SourceUpstream != "upstream-b" { + t.Fatalf("selected upstream = %q, want upstream-b", lease.Proxy.SourceUpstream) + } + if err := lease.Cancel(); err != nil { + t.Fatalf("Cancel(): %v", err) + } +} + func TestAcquireSurvivesConcurrentSnapshotApply(t *testing.T) { now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) store := snapshot.NewStore("cluster-a", "worker-a") @@ -246,6 +426,40 @@ func BenchmarkAcquire100kIndexed(b *testing.B) { } } +func BenchmarkAcquire100kRoutingRoundRobin(b *testing.B) { + store := snapshot.NewStore("cluster-a", "worker-a") + proxies := makeTestProxies(100_000) + envelope := snapshot.Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, Proxies: proxies, + } + envelope.Checksum = snapshot.Checksum(envelope.Proxies) + if err := store.Apply(envelope); err != nil { + b.Fatalf("Apply(): %v", err) + } + + dispatcher := New(store) + request := Request{ + Now: time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC), RoutingName: "benchmark-round-robin", + Upstreams: []string{"upstream-a", "upstream-b"}, Strategy: routing.Strategy{Type: routing.StrategyRoundRobin}, + Scheme: proxyDomain.SchemeHTTP, RequiredTags: map[string]string{"region": "cn-east", "tier": "gold"}, + SafetyMargin: 5 * time.Second, + } + + b.ReportAllocs() + b.ResetTimer() + b.RunParallel(func(parallel *testing.PB) { + for parallel.Next() { + lease, err := dispatcher.Acquire(request) + if err != nil { + b.Fatalf("Acquire(): %v", err) + } + if err := lease.Cancel(); err != nil { + b.Fatalf("Cancel(): %v", err) + } + } + }) +} + func makeTestProxies(count int) []proxyDomain.Proxy { now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) expiresAt := now.Add(10 * time.Minute) @@ -269,6 +483,34 @@ func makeTestProxies(count int) []proxyDomain.Proxy { return proxies } +func routingStrategyStore(t *testing.T) *snapshot.Store { + t.Helper() + store := snapshot.NewStore("cluster-a", "worker-a") + envelope := snapshot.Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + Proxies: []proxyDomain.Proxy{ + {ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, SourceUpstream: "upstream-a", State: proxyDomain.StateAvailable, MaxConcurrency: 1}, + {ID: "proxy-b", Scheme: proxyDomain.SchemeHTTP, SourceUpstream: "upstream-b", State: proxyDomain.StateAvailable, MaxConcurrency: 1}, + }, + } + envelope.Checksum = snapshot.Checksum(envelope.Proxies) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply(): %v", err) + } + return store +} + +type fixedRoutingRandom struct { + values []int + index int +} + +func (source *fixedRoutingRandom) Intn(limit int) int { + value := source.values[source.index] + source.index++ + return value % limit +} + func schemeForIndex(index int) proxyDomain.Scheme { switch index % 3 { case 0: diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index 608ddc8..3ec89d1 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -338,6 +338,30 @@ func (v *View) MatchRouting(request routing.Request) (routing.Rule, bool) { return v.routing.Match(request) } +// UpstreamLoad returns the current in-flight load and whether an eligible +// capacity slot remains for one Upstream under the supplied request filters. +func (v *View) UpstreamLoad(query Query, upstream string) (int64, bool) { + if v == nil || upstream == "" { + return 0, false + } + query.Upstreams = []string{upstream} + selection := v.Select(query) + var active int64 + available := false + for index := 0; index < selection.Len(); index++ { + entry, ok := selection.EntryAt(index) + if !ok { + continue + } + entryActive, reserved, maximum := entry.Runtime.Counters() + active += entryActive + reserved + if entryActive+reserved < maximum { + available = true + } + } + return active, available +} + func (s *Store) reclaimableRuntimes(nextProxyIDs map[string]struct{}) map[string]*runtimeRegistration { result := make(map[string]*runtimeRegistration) for proxyID, registration := range s.runtimes {