From bc94f4dbf1e88de0e5f7ea297be848efa8bfb528 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 15:29:10 +0800 Subject: [PATCH] feat: publish gateway routing snapshots --- README.md | 9 +- docs/api/control-plane.md | 8 +- docs/development/implementation-plan.md | 5 +- docs/requirements/completion-audit.md | 9 +- docs/requirements/traceability.md | 2 +- internal/config/store.go | 14 +- internal/config/store_test.go | 19 ++ internal/controller/bootstrap/bootstrap.go | 12 + .../worker/owned_snapshot_source.go | 30 ++- .../worker/routing_snapshot_source.go | 249 ++++++++++++++++++ .../worker/routing_snapshot_source_test.go | 131 +++++++++ internal/controller/worker/service.go | 7 +- .../controller/worker/snapshot_source_test.go | 44 ++++ 13 files changed, 518 insertions(+), 21 deletions(-) create mode 100644 internal/controller/worker/routing_snapshot_source.go create mode 100644 internal/controller/worker/routing_snapshot_source_test.go diff --git a/README.md b/README.md index 31c8359..0361304 100644 --- a/README.md +++ b/README.md @@ -43,8 +43,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 - **Gateway 组件**：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、 目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/ Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行 - 有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 快照； - `proxy-gateway` 命令、凭据分发和 Routing Snapshot 发布仍待装配。 + 有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 及 Gateway + Routing 快照；`proxy-gateway` 命令、凭据分发和 Routing Snapshot 消费仍待装配。 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 @@ -83,8 +83,9 @@ flowchart LR Redis 会话栅栏。 - **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership 与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。 -- **待完成**：带凭据 Proxy 与 Routing 的权威 Snapshot 发布、Gateway 进程装配、Outcome - 上报、Checker 调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性集群压测。 +- **待完成**：带凭据 Proxy 分发、Gateway Routing Snapshot 消费、Gateway 进程装配、 + Outcome 上报、Checker 调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性 + 集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 58f94c6..e856eb3 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -28,9 +28,11 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和 `WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照， Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership 索引构建无凭据引用的已归属 Proxy 内容，并将租约到期收紧到 Proxy 的 -`usable_until`；带凭据 Proxy 会 fail-closed，直到凭据材料分发完成。Routing 内容、 -增量、Gateway 进程装配、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确 -返回 `Unimplemented`；`100,000 QPS` 仍是未验证的设计目标。 +`usable_until`；带凭据 Proxy 会 fail-closed，直到凭据材料分发完成。完整 Snapshot +已从配置原始顺序和 Admin 当前状态合成 Gateway Routing，并与 Proxy 一起纳入 checksum； +Gateway 对该 Routing payload 的运行时消费、增量、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 156e9a2..ee64594 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -217,8 +217,9 @@ Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话 已归属 Proxy payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成， 重注册会清除旧引用，避免迟到 Stream 覆盖新 session。Worker 服务端会在最近完整 Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已为 Gateway 调用方 -提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Routing payload、 -凭据分发、Gateway 命令与 Outcome 上报仍未实现。 +提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Gateway Routing +payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 checksum；其 Gateway +运行时消费、凭据分发、Gateway 命令与 Outcome 上报仍未实现。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 9b3b57c..53eebd1 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -51,7 +51,8 @@ 服务端 TTL，负向 ACK 会关闭 Runtime 写入栅栏。Controller 已装配 loopback plaintext fixture 与 SPIFFE mTLS 服务端；基础 Snapshot 流和 Gateway 的 Register/Watch/ACK/Runtime 会话协调已实现。Redis 以 Worker 可下发 ownership - 索引构建无凭据引用的已归属 Proxy payload，并以租约收紧可用期；Routing payload、 + 索引构建无凭据引用的已归属 Proxy payload，并以租约收紧可用期；Gateway Routing + payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum，其运行时消费、 凭据分发、Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用 @@ -98,9 +99,9 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt 5. Worker Register/ACK/Runtime gRPC 接收路径、Redis 服务端 TTL 会话生命周期、 SPIFFE mTLS 校验和 Controller 生命周期接线已完成；Redis Provider Leader、 分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。 -6. Worker 基础网络快照流、无凭据引用 Proxy payload 和 Gateway Snapshot 客户端已 - 完成；Routing payload、凭据分发与 Outcome 上报仍待完成。Redis ownership - drain/ACK/过期回收及按 Worker 的可下发索引已完成。 +6. Worker 基础网络快照流、无凭据引用 Proxy/Gateway Routing payload 和 Gateway + Snapshot 客户端已完成；Gateway Routing 运行时消费、凭据分发与 Outcome 上报仍待 + 完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 7. Checker 调度、探测器和健康 reducer。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 0ba234b..e29dbda 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -7,7 +7,7 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，Gateway 已有会话客户端组件；Gateway/Checker/Loadgen 构建产物及权威 Snapshot payload 发布待实现 | +| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，并发布无凭据 Proxy/Gateway Routing 完整快照，Gateway 已有会话客户端组件；Gateway/Checker/Loadgen 构建产物及 Gateway Routing 运行时消费待实现 | | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Snapshot/Dispatch 及依赖边界已验证；完整 Gateway 进程与代表性性能剖析待完成 | | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 生产入口待装配 | | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | diff --git a/internal/config/store.go b/internal/config/store.go index e3c5ae6..2858cbb 100644 --- a/internal/config/store.go +++ b/internal/config/store.go @@ -27,15 +27,23 @@ func NewStore(initial *Config) (*Store, error) { } func (store *Store) Current() *Config { + configuration, _ := store.Snapshot() + return configuration +} + +// Snapshot returns a detached configuration and its revision from the same +// atomic publication. Callers that depend on both must not read them +// separately through Current and Revision. +func (store *Store) Snapshot() (*Config, uint64) { if store == nil { - return nil + return nil, 0 } published := store.current.Load() if published == nil { - return nil + return nil, 0 } cloned := cloneConfig(published.value) - return &cloned + return &cloned, published.revision } func (store *Store) Revision() uint64 { diff --git a/internal/config/store_test.go b/internal/config/store_test.go index 9cb2bde..ea96a31 100644 --- a/internal/config/store_test.go +++ b/internal/config/store_test.go @@ -94,6 +94,25 @@ func TestStoreRejectsOutOfOrderRevisionPublication(t *testing.T) { } } +func TestStoreSnapshotsConfigurationAndRevisionFromOnePublication(t *testing.T) { + t.Parallel() + store, err := NewStore(storeTestConfig("provider-a")) + if err != nil { + t.Fatalf("NewStore(): %v", err) + } + if !store.PublishRevision(storeTestConfig("provider-b"), 7) { + t.Fatal("PublishRevision() = false") + } + configuration, revision := store.Snapshot() + if configuration == nil || revision != 7 || configuration.Routing[0].Upstreams[0] != "provider-b" { + t.Fatalf("Snapshot() = %+v, %d", configuration, revision) + } + configuration.Routing[0].Upstreams[0] = "mutated" + if current, _ := store.Snapshot(); current.Routing[0].Upstreams[0] != "provider-b" { + t.Fatalf("Snapshot() exposed mutable publication: %+v", current) + } +} + func storeTestConfig(upstreamName string) *Config { return &Config{ Version: 1, diff --git a/internal/controller/bootstrap/bootstrap.go b/internal/controller/bootstrap/bootstrap.go index a26cf89..2075a83 100644 --- a/internal/controller/bootstrap/bootstrap.go +++ b/internal/controller/bootstrap/bootstrap.go @@ -233,6 +233,17 @@ func runWithWorkerFactory( if nilInterface(opened.workerStore) { return errors.Join(ErrStartup, ErrInvalidOptions) } + var routingSources []worker.RoutingStateReader + if loaded.Value.Admin.Enabled { + if nilInterface(opened.state) { + return errors.Join(ErrStartup, ErrInvalidOptions) + } + routingSources = append(routingSources, opened.state) + } + routingSource, routingErr := worker.NewGatewayRoutingSource(configurationStore, routingSources...) + if routingErr != nil { + return fmt.Errorf("%w: build Worker routing source: %w", ErrStartup, routingErr) + } var snapshotReader ownershipDomain.SnapshotReader if reader, ok := opened.workerStore.(ownershipDomain.SnapshotReader); ok { snapshotReader = reader @@ -245,6 +256,7 @@ func runWithWorkerFactory( MaxRuntimeCounters: loaded.Value.ControlPlane.MaxRuntimeCounters, MaxSnapshotBytes: loaded.Value.ControlPlane.MaxMessageBytes, SnapshotReader: snapshotReader, + RoutingSource: routingSource, }) if serviceErr != nil { return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr) diff --git a/internal/controller/worker/owned_snapshot_source.go b/internal/controller/worker/owned_snapshot_source.go index 3ac01f2..8aa13ea 100644 --- a/internal/controller/worker/owned_snapshot_source.go +++ b/internal/controller/worker/owned_snapshot_source.go @@ -27,6 +27,7 @@ type OwnedSnapshotSource struct { maxProxies int maxBytes int now func() time.Time + routing RoutingSource } func NewOwnedSnapshotSource( @@ -36,13 +37,19 @@ func NewOwnedSnapshotSource( maxProxies int, maxBytes int, now func() time.Time, + routing ...RoutingSource, ) (*OwnedSnapshotSource, error) { - if epochs == nil || reader == nil || validFor <= 0 || maxProxies <= 0 || maxBytes <= 0 || now == nil { + if epochs == nil || reader == nil || validFor <= 0 || maxProxies <= 0 || maxBytes <= 0 || now == nil || len(routing) > 1 || + len(routing) == 1 && routing[0] == nil { return nil, ErrSnapshotsUnavailable } - return &OwnedSnapshotSource{ + source := &OwnedSnapshotSource{ epochs: epochs, reader: reader, validFor: validFor, maxProxies: maxProxies, maxBytes: maxBytes, now: now, - }, nil + } + if len(routing) == 1 { + source.routing = routing[0] + } + return source, nil } func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWatchRequest) (<-chan *controlplanev1.WorkerSnapshot, error) { @@ -80,6 +87,13 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa } full.Proxies = append(full.Proxies, owned) } + if source.routing != nil { + routing, err := source.routing.Read(ctx) + if err != nil { + return nil, err + } + full.Routing = cloneSnapshotRouting(routing) + } checksum, err := snapshotwire.Checksum(full) if err != nil { return nil, err @@ -93,6 +107,16 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa return updates, nil } +func cloneSnapshotRouting(source []*controlplanev1.RoutingRule) []*controlplanev1.RoutingRule { + result := make([]*controlplanev1.RoutingRule, len(source)) + for index, rule := range source { + if rule != nil { + result[index] = proto.Clone(rule).(*controlplanev1.RoutingRule) + } + } + return result +} + func wireOwnedProxy(item ownershipDomain.SnapshotProxy, now time.Time) (*controlplanev1.OwnedProxy, time.Time, error) { proxy := item.Proxy if item.OwnershipEpoch == 0 || item.LeaseExpiresAt.IsZero() || proxy.CredentialVersion != "" || proxy.SecretRef != "" { diff --git a/internal/controller/worker/routing_snapshot_source.go b/internal/controller/worker/routing_snapshot_source.go new file mode 100644 index 0000000..9d8f959 --- /dev/null +++ b/internal/controller/worker/routing_snapshot_source.go @@ -0,0 +1,249 @@ +package worker + +import ( + "context" + "errors" + "fmt" + "math" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" +) + +var ErrRoutingUnavailable = errors.New("worker snapshot routing is unavailable") + +// RoutingSource reads the complete, ordered Gateway routing payload for a +// WorkerSnapshot. It never performs data-plane reads. +type RoutingSource interface { + Read(context.Context) ([]*controlplanev1.RoutingRule, error) +} + +type RoutingConfigurationReader interface { + Snapshot() (*config.Config, uint64) +} + +type RoutingStateReader interface { + Snapshot(context.Context) (adminstate.Snapshot, error) +} + +// GatewayRoutingSource combines immutable configuration semantics with the +// current Admin routing and Upstream state. Admin state is optional only for +// deployments that do not enable the Admin control surface. +type GatewayRoutingSource struct { + configuration RoutingConfigurationReader + state RoutingStateReader +} + +func NewGatewayRoutingSource(configuration RoutingConfigurationReader, states ...RoutingStateReader) (*GatewayRoutingSource, error) { + if configuration == nil || len(states) > 1 || len(states) == 1 && states[0] == nil { + return nil, ErrRoutingUnavailable + } + source := &GatewayRoutingSource{configuration: configuration} + if len(states) == 1 { + source.state = states[0] + } + return source, nil +} + +func (source *GatewayRoutingSource) Read(ctx context.Context) ([]*controlplanev1.RoutingRule, error) { + if source == nil || source.configuration == nil || ctx == nil { + return nil, ErrRoutingUnavailable + } + if err := ctx.Err(); err != nil { + return nil, err + } + configuration, revision := source.configuration.Snapshot() + if configuration == nil { + return nil, ErrRoutingUnavailable + } + var state *adminstate.Snapshot + if source.state != nil { + current, err := source.state.Snapshot(ctx) + if err != nil { + return nil, errors.Join(ErrRoutingUnavailable, err) + } + if current.Config == nil || current.Config.Revision != revision { + return nil, ErrRoutingUnavailable + } + state = ¤t + } + rules, err := buildGatewayRouting(configuration, state) + if err != nil { + return nil, errors.Join(ErrRoutingUnavailable, err) + } + return rules, nil +} + +func buildGatewayRouting(configuration *config.Config, state *adminstate.Snapshot) ([]*controlplanev1.RoutingRule, error) { + if configuration == nil { + return nil, errors.New("configuration is nil") + } + upstreamEnabled, err := enabledGatewayUpstreams(configuration, state) + if err != nil { + return nil, err + } + routingStates, err := gatewayRoutingStates(configuration, state) + if err != nil { + return nil, err + } + rules := make([]*controlplanev1.RoutingRule, 0, len(configuration.Routing)) + for _, configured := range configuration.Routing { + if configured.Purpose != "gateway" { + continue + } + state, hasState := routingStates[configured.Name] + if !hasState && routingStates != nil { + return nil, fmt.Errorf("routing %q is missing from admin state", configured.Name) + } + upstreams := enabledRoutingUpstreams(configured.Upstreams, upstreamEnabled) + current := configuredCurrentUpstream(configured, state, hasState) + strategy, err := wireRoutingStrategy(configured.Strategy, current, upstreams) + if err != nil { + return nil, fmt.Errorf("routing %q strategy: %w", configured.Name, err) + } + enabled := configured.Enabled && (routingStates == nil || state.Enabled) && len(upstreams) > 0 + if strategy.GetType() == controlplanev1.StrategyType_STRATEGY_TYPE_SEQUENTIAL && !containsUpstream(upstreams, current) { + enabled = false + } + action, err := wireUnavailableAction(configured.OnUnavailable.Action) + if err != nil { + return nil, fmt.Errorf("routing %q unavailable action: %w", configured.Name, err) + } + rules = append(rules, &controlplanev1.RoutingRule{ + Name: configured.Name, Enabled: enabled, HostRegex: configured.Match.HostRegex, + Methods: append([]string(nil), configured.Match.Methods...), PathRegex: configured.Match.PathRegex, + Headers: cloneRoutingHeaders(configured.Match.Headers), Upstreams: upstreams, Strategy: strategy, + OnUnavailable: action, + }) + } + return rules, nil +} + +func enabledGatewayUpstreams(configuration *config.Config, state *adminstate.Snapshot) (map[string]bool, error) { + adminStates := make(map[string]adminstate.UpstreamState) + if state != nil { + for _, upstream := range state.Upstreams { + if _, duplicate := adminStates[upstream.Name]; duplicate { + return nil, fmt.Errorf("duplicate upstream state %q", upstream.Name) + } + adminStates[upstream.Name] = upstream + } + } + result := make(map[string]bool, len(configuration.Upstreams)) + for name, upstream := range configuration.Upstreams { + result[name] = upstream.Enabled + if state == nil { + continue + } + admin, exists := adminStates[name] + if !exists { + return nil, fmt.Errorf("upstream %q is missing from admin state", name) + } + result[name] = result[name] && admin.Enabled + } + return result, nil +} + +func gatewayRoutingStates(configuration *config.Config, state *adminstate.Snapshot) (map[string]adminstate.RoutingState, error) { + if state == nil { + return nil, nil + } + result := make(map[string]adminstate.RoutingState, len(state.Routings)) + for _, routing := range state.Routings { + if _, duplicate := result[routing.Name]; duplicate { + return nil, fmt.Errorf("duplicate routing state %q", routing.Name) + } + result[routing.Name] = routing + } + for _, routing := range configuration.Routing { + if routing.Purpose == "gateway" { + if _, exists := result[routing.Name]; !exists { + return nil, fmt.Errorf("routing %q is missing from admin state", routing.Name) + } + } + } + return result, nil +} + +func enabledRoutingUpstreams(source []string, enabled map[string]bool) []string { + result := make([]string, 0, len(source)) + for _, upstream := range source { + if enabled[upstream] { + result = append(result, upstream) + } + } + return result +} + +func configuredCurrentUpstream(routing config.Routing, state adminstate.RoutingState, hasState bool) string { + if hasState { + return state.CurrentUpstream + } + if len(routing.Upstreams) == 0 { + return "" + } + return routing.Upstreams[0] +} + +func wireRoutingStrategy(configured config.Strategy, current string, upstreams []string) (*controlplanev1.RoutingStrategy, error) { + strategy := &controlplanev1.RoutingStrategy{CurrentUpstream: current} + switch configured.Type { + case "sequential": + strategy.Type = controlplanev1.StrategyType_STRATEGY_TYPE_SEQUENTIAL + case "random": + strategy.Type = controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM + case "roundRobin": + strategy.Type = controlplanev1.StrategyType_STRATEGY_TYPE_ROUND_ROBIN + case "weighted": + strategy.Type = controlplanev1.StrategyType_STRATEGY_TYPE_WEIGHTED + strategy.Weights = make(map[string]uint32, len(upstreams)) + for _, upstream := range upstreams { + weight, exists := configured.Weights[upstream] + if !exists || weight <= 0 || weight > math.MaxUint32 { + return nil, fmt.Errorf("invalid weight for upstream %q", upstream) + } + strategy.Weights[upstream] = uint32(weight) + } + case "leastConnections": + strategy.Type = controlplanev1.StrategyType_STRATEGY_TYPE_LEAST_CONNECTIONS + default: + return nil, fmt.Errorf("unsupported strategy type %q", configured.Type) + } + return strategy, nil +} + +func wireUnavailableAction(action string) (controlplanev1.UnavailableAction, error) { + switch action { + case "reject": + return controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_REJECT, nil + case "wait": + return controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT, nil + case "direct": + return controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_DIRECT, nil + default: + return controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_UNSPECIFIED, fmt.Errorf("unsupported action %q", action) + } +} + +func cloneRoutingHeaders(source map[string]string) map[string]string { + if source == nil { + return nil + } + result := make(map[string]string, len(source)) + for name, value := range source { + result[name] = value + } + return result +} + +func containsUpstream(upstreams []string, target string) bool { + for _, upstream := range upstreams { + if upstream == target { + return true + } + } + return false +} + +var _ RoutingSource = (*GatewayRoutingSource)(nil) diff --git a/internal/controller/worker/routing_snapshot_source_test.go b/internal/controller/worker/routing_snapshot_source_test.go new file mode 100644 index 0000000..cc2c297 --- /dev/null +++ b/internal/controller/worker/routing_snapshot_source_test.go @@ -0,0 +1,131 @@ +package worker + +import ( + "context" + "errors" + "reflect" + "testing" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/config" + "proxy-pool/internal/domain/adminstate" +) + +func TestGatewayRoutingSourceBuildsOrderedRulesFromConfigurationAndAdminState(t *testing.T) { + configuration := &config.Config{ + Routing: []config.Routing{ + {Name: "extract", Enabled: true, Purpose: "extract", Upstreams: []string{"provider-a"}, Strategy: config.Strategy{Type: "random"}, OnUnavailable: config.OnUnavailable{Action: "reject"}}, + { + Name: "checkout", Enabled: true, Purpose: "gateway", + Match: config.RoutingMatch{HostRegex: "^api\\.example$", Methods: []string{"GET"}, Headers: map[string]string{"X-Tier": "gold"}}, + Upstreams: []string{"provider-a", "provider-b"}, Strategy: config.Strategy{Type: "sequential", SwitchAfterEmptyFetch: 5}, + OnUnavailable: config.OnUnavailable{Action: "reject"}, + }, + { + Name: "fallback", Enabled: true, Purpose: "gateway", Upstreams: []string{"provider-a", "provider-c"}, + Strategy: config.Strategy{Type: "weighted", Weights: map[string]int{"provider-a": 3, "provider-c": 7}}, + OnUnavailable: config.OnUnavailable{Action: "wait"}, + }, + }, + Upstreams: map[string]config.Upstream{ + "provider-a": {Enabled: true}, "provider-b": {Enabled: false}, "provider-c": {Enabled: true}, + }, + } + state := adminstate.Snapshot{ + Config: &adminstate.ConfigRevision{Revision: 9}, + Upstreams: []adminstate.UpstreamState{ + {Name: "provider-a", Enabled: true}, {Name: "provider-b", Enabled: false}, {Name: "provider-c", Enabled: true}, + }, + Routings: []adminstate.RoutingState{ + {Name: "checkout", Enabled: true, Upstreams: []string{"provider-a", "provider-b"}, CurrentUpstream: "provider-a"}, + {Name: "fallback", Enabled: true, Upstreams: []string{"provider-a", "provider-c"}, CurrentUpstream: "provider-a"}, + }, + } + source, err := NewGatewayRoutingSource(staticGatewayRoutingConfiguration{configuration: configuration, revision: 9}, staticGatewayRoutingState{snapshot: state}) + 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) != 2 || rules[0].GetName() != "checkout" || rules[1].GetName() != "fallback" { + t.Fatalf("rules = %+v, want ordered gateway rules", rules) + } + checkout := rules[0] + if !checkout.GetEnabled() || checkout.GetStrategy().GetType() != controlplanev1.StrategyType_STRATEGY_TYPE_SEQUENTIAL || + checkout.GetStrategy().GetCurrentUpstream() != "provider-a" || !reflect.DeepEqual(checkout.GetUpstreams(), []string{"provider-a"}) || + checkout.GetHostRegex() != "^api\\.example$" || checkout.GetHeaders()["X-Tier"] != "gold" { + t.Fatalf("checkout rule = %+v", checkout) + } + fallback := rules[1] + if !fallback.GetEnabled() || fallback.GetOnUnavailable() != controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT || + !reflect.DeepEqual(fallback.GetUpstreams(), []string{"provider-a", "provider-c"}) || + !reflect.DeepEqual(fallback.GetStrategy().GetWeights(), map[string]uint32{"provider-a": 3, "provider-c": 7}) { + t.Fatalf("fallback rule = %+v", fallback) + } +} + +func TestGatewayRoutingSourceFailsClosedWhenAdminStateDoesNotMatchConfiguration(t *testing.T) { + configuration := &config.Config{Routing: []config.Routing{{ + Name: "gateway", Enabled: true, Purpose: "gateway", Upstreams: []string{"provider-a"}, + Strategy: config.Strategy{Type: "random"}, OnUnavailable: config.OnUnavailable{Action: "reject"}, + }}, Upstreams: map[string]config.Upstream{"provider-a": {Enabled: true}}} + source, err := NewGatewayRoutingSource( + staticGatewayRoutingConfiguration{configuration: configuration, revision: 5}, + staticGatewayRoutingState{snapshot: adminstate.Snapshot{Config: &adminstate.ConfigRevision{Revision: 4}}}, + ) + if err != nil { + t.Fatalf("NewGatewayRoutingSource(): %v", err) + } + if _, err := source.Read(context.Background()); !errors.Is(err, ErrRoutingUnavailable) { + t.Fatalf("Read() error = %v, want ErrRoutingUnavailable", err) + } +} + +func TestGatewayRoutingSourceDisablesSequentialRuleWhenCurrentUpstreamIsUnavailable(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-a"}}, + }}, + ) + 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-a" || + !reflect.DeepEqual(rules[0].GetUpstreams(), []string{"provider-b"}) { + t.Fatalf("rules = %+v, want disabled sequential rule with provider-b remaining", rules) + } +} + +type staticGatewayRoutingConfiguration struct { + configuration *config.Config + revision uint64 +} + +func (reader staticGatewayRoutingConfiguration) Snapshot() (*config.Config, uint64) { + return reader.configuration, reader.revision +} + +type staticGatewayRoutingState struct { + snapshot adminstate.Snapshot + err error +} + +func (reader staticGatewayRoutingState) Snapshot(context.Context) (adminstate.Snapshot, error) { + return reader.snapshot, reader.err +} diff --git a/internal/controller/worker/service.go b/internal/controller/worker/service.go index f2efdd3..135217d 100644 --- a/internal/controller/worker/service.go +++ b/internal/controller/worker/service.go @@ -58,6 +58,7 @@ type Options struct { MaxRuntimeCounters int MaxSnapshotBytes int SnapshotReader ownershipDomain.SnapshotReader + RoutingSource RoutingSource SessionID func() (string, error) } @@ -131,8 +132,12 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err } result := &service{store: store, options: options} if options.SnapshotReader != nil { + var routing []RoutingSource + if options.RoutingSource != nil { + routing = append(routing, options.RoutingSource) + } source, err := NewOwnedSnapshotSource( - result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, + result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, routing..., ) if err != nil { return nil, errors.Join(ErrInvalidCommand, err) diff --git a/internal/controller/worker/snapshot_source_test.go b/internal/controller/worker/snapshot_source_test.go index ca6b279..0a45a25 100644 --- a/internal/controller/worker/snapshot_source_test.go +++ b/internal/controller/worker/snapshot_source_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/controlplane/snapshotwire" ownershipDomain "proxy-pool/internal/domain/ownership" proxyDomain "proxy-pool/internal/domain/proxy" @@ -67,6 +68,40 @@ func TestOwnedSnapshotSourceBuildsBoundedProxySnapshot(t *testing.T) { } } +func TestOwnedSnapshotSourceIncludesRoutingInFullSnapshotChecksum(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + expiresAt := now.Add(10 * time.Minute) + usableUntil := now.Add(5 * time.Minute) + routing := &controlplanev1.RoutingRule{ + Name: "gateway", Enabled: true, HostRegex: ".*", Upstreams: []string{"upstream-a"}, + Strategy: &controlplanev1.RoutingStrategy{Type: controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM}, + OnUnavailable: controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_REJECT, + } + source, err := NewOwnedSnapshotSource(epochReaderStub{epoch: 9}, snapshotReaderStub{proxies: []ownershipDomain.SnapshotProxy{{ + Proxy: proxyDomain.Proxy{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, + SourceUpstream: "upstream-a", ExpiresAt: &expiresAt, UsableUntil: &usableUntil, + MaxConcurrency: 7, State: proxyDomain.StateAvailable, + }, + OwnershipEpoch: 4, LeaseExpiresAt: now.Add(time.Minute), + }}}, time.Minute*2, 10, 4096, func() time.Time { return now }, routingSourceStub{rules: []*controlplanev1.RoutingRule{routing}}) + if err != nil { + t.Fatalf("NewOwnedSnapshotSource(): %v", err) + } + updates, err := source.Watch(context.Background(), SnapshotWatchRequest{WorkerID: "worker-a", SessionID: "session-a"}) + if err != nil { + t.Fatalf("Watch(): %v", err) + } + full := <-updates + if len(full.GetRouting()) != 1 || full.GetRouting()[0].GetName() != "gateway" || full.GetRouting()[0] == routing { + t.Fatalf("snapshot routing = %+v", full.GetRouting()) + } + checksum, err := snapshotwire.Checksum(full) + if err != nil || string(checksum[:]) != string(full.GetChecksum()) { + t.Fatalf("snapshot checksum = %x, %v; want %x", full.GetChecksum(), err, checksum) + } +} + func TestOwnedSnapshotSourceRejectsCredentialReferenceUntilMaterialIsAvailable(t *testing.T) { now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) expiresAt := now.Add(time.Minute) @@ -102,6 +137,15 @@ type snapshotReaderStub struct { err error } +type routingSourceStub struct { + rules []*controlplanev1.RoutingRule + err error +} + +func (source routingSourceStub) Read(context.Context) ([]*controlplanev1.RoutingRule, error) { + return source.rules, source.err +} + func (reader snapshotReaderStub) ReadWorkerSnapshot(context.Context, string, int) ([]ownershipDomain.SnapshotProxy, error) { return reader.proxies, reader.err }