diff --git a/README.md b/README.md index 0361304..e4401c7 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,9 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/ Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行 有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 及 Gateway - Routing 快照；`proxy-gateway` 命令、凭据分发和 Routing Snapshot 消费仍待装配。 + Routing 快照。Gateway 会将 Routing 与 Proxy 原子编译为同一内存 View，并只按当前未过期 + View 匹配请求；`proxy-gateway` 命令、凭据分发，以及策略选择与 `onUnavailable` 派发语义 + 仍待装配。 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 @@ -83,8 +85,8 @@ flowchart LR Redis 会话栅栏。 - **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership 与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。 -- **待完成**：带凭据 Proxy 分发、Gateway Routing Snapshot 消费、Gateway 进程装配、 - Outcome 上报、Checker 调度与健康状态链、完整 Routing 运行链，以及 loadgen 和代表性 +- **待完成**：带凭据 Proxy 分发、Gateway 进程装配、Routing 策略选择与 + `onUnavailable`、Outcome 上报、Checker 调度与健康状态链，以及 loadgen 和代表性 集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index e856eb3..af4415a 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -30,9 +30,9 @@ Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有 索引构建无凭据引用的已归属 Proxy 内容，并将租约到期收紧到 Proxy 的 `usable_until`；带凭据 Proxy 会 fail-closed，直到凭据材料分发完成。完整 Snapshot 已从配置原始顺序和 Admin 当前状态合成 Gateway Routing，并与 Proxy 一起纳入 checksum； -Gateway 对该 Routing payload 的运行时消费、增量、Gateway 进程装配、Outcome 与 Checker -闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；`100,000 QPS` 仍是未验证的 -设计目标。 +Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本地 View，动态 Router 只匹配 +当前未过期 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 ee64594..f3d7b01 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -218,8 +218,9 @@ Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话 重注册会清除旧引用，避免迟到 Stream 覆盖新 session。Worker 服务端会在最近完整 Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已为 Gateway 调用方 提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Gateway Routing -payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 checksum；其 Gateway -运行时消费、凭据分发、Gateway 命令与 Outcome 上报仍未实现。 +payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 checksum；Gateway 已将其与 +Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。策略选择、`onUnavailable`、 +凭据分发、Gateway 命令与 Outcome 上报仍未实现。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 53eebd1..6f4fee8 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -99,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/Gateway Routing payload 和 Gateway - Snapshot 客户端已完成；Gateway Routing 运行时消费、凭据分发与 Outcome 上报仍待 - 完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 +6. Worker 基础网络快照流、无凭据引用 Proxy/Gateway Routing payload、Gateway Snapshot + 客户端及同版本 Routing 编译/动态匹配已完成；策略选择、`onUnavailable`、凭据分发与 + 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 e29dbda..34a13cc 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，并发布无凭据 Proxy/Gateway Routing 完整快照，Gateway 已有会话客户端组件；Gateway/Checker/Loadgen 构建产物及 Gateway Routing 运行时消费待实现 | +| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，并发布无凭据 Proxy/Gateway Routing 完整快照；Gateway 已将同版本 Routing 编译为动态匹配 View。Gateway/Checker/Loadgen 构建产物、策略派发与 `onUnavailable` 待实现 | | 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/domain/routing/rule.go b/internal/domain/routing/rule.go index fdfc290..4eae180 100644 --- a/internal/domain/routing/rule.go +++ b/internal/domain/routing/rule.go @@ -15,6 +15,30 @@ const ( ActionReject Action = "reject" ) +type StrategyType string + +const ( + StrategySequential StrategyType = "sequential" + StrategyRandom StrategyType = "random" + StrategyRoundRobin StrategyType = "roundRobin" + StrategyWeighted StrategyType = "weighted" + StrategyLeastConnections StrategyType = "leastConnections" +) + +type Strategy struct { + Type StrategyType + CurrentUpstream string + Weights map[string]uint32 +} + +type OnUnavailableAction string + +const ( + OnUnavailableReject OnUnavailableAction = "reject" + OnUnavailableWait OnUnavailableAction = "wait" + OnUnavailableDirect OnUnavailableAction = "direct" +) + type Match struct { HostRegex string Methods []string @@ -23,10 +47,12 @@ type Match struct { } type Rule struct { - Name string - Match Match - Upstreams []string - Action Action + Name string + Match Match + Upstreams []string + Action Action + Strategy Strategy + OnUnavailable OnUnavailableAction } type Request struct { @@ -106,6 +132,12 @@ func cloneRule(source Rule) Rule { cloned.Match.Headers[name] = value } } + if source.Strategy.Weights != nil { + cloned.Strategy.Weights = make(map[string]uint32, len(source.Strategy.Weights)) + for upstream, weight := range source.Strategy.Weights { + cloned.Strategy.Weights[upstream] = weight + } + } return cloned } diff --git a/internal/gateway/controlplane/watcher.go b/internal/gateway/controlplane/watcher.go index 5a6b6bd..bde721a 100644 --- a/internal/gateway/controlplane/watcher.go +++ b/internal/gateway/controlplane/watcher.go @@ -4,6 +4,7 @@ import ( "context" "crypto/sha256" "errors" + "fmt" "io" "math" "time" @@ -11,6 +12,7 @@ import ( controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/controlplane/snapshotwire" proxyDomain "proxy-pool/internal/domain/proxy" + routingDomain "proxy-pool/internal/domain/routing" "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/gateway/snapshot" @@ -123,14 +125,147 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e if err != nil { return err } + routes, err := wireRouting(full.GetRouting()) + if err != nil { + return err + } envelope := snapshot.Envelope{ ClusterID: watcher.options.ClusterID, WorkerID: watcher.options.WorkerID, - Epoch: full.GetOwnershipEpoch(), Version: full.GetVersion(), Full: true, ValidUntil: validUntil, Proxies: proxies, + Epoch: full.GetOwnershipEpoch(), Version: full.GetVersion(), Full: true, ValidUntil: validUntil, + Proxies: proxies, Routing: routes, } - envelope.Checksum = snapshot.Checksum(proxies) + envelope.Checksum = snapshot.ChecksumWithRouting(proxies, routes) return watcher.store.Apply(envelope) } +func wireRouting(source []*controlplanev1.RoutingRule) ([]routingDomain.Rule, error) { + result := make([]routingDomain.Rule, 0, len(source)) + names := make(map[string]struct{}, len(source)) + for _, rule := range source { + if rule == nil || !workerruntime.ValidIdentifier(rule.GetName()) { + return nil, ErrInvalidSnapshotWatcher + } + if _, duplicate := names[rule.GetName()]; duplicate { + return nil, ErrInvalidSnapshotWatcher + } + names[rule.GetName()] = struct{}{} + if !rule.GetEnabled() { + continue + } + upstreams, err := wireRoutingUpstreams(rule.GetUpstreams()) + if err != nil { + return nil, err + } + strategy, err := wireRoutingStrategy(rule.GetStrategy(), upstreams) + if err != nil { + return nil, err + } + action, err := wireUnavailableAction(rule.GetOnUnavailable()) + if err != nil { + return nil, err + } + result = append(result, routingDomain.Rule{ + Name: rule.GetName(), + Match: routingDomain.Match{ + HostRegex: rule.GetHostRegex(), Methods: append([]string(nil), rule.GetMethods()...), + PathRegex: rule.GetPathRegex(), Headers: cloneRoutingHeaders(rule.GetHeaders()), + }, + Upstreams: upstreams, Action: routingDomain.ActionProxy, Strategy: strategy, OnUnavailable: action, + }) + } + return result, nil +} + +func wireRoutingUpstreams(source []string) ([]string, error) { + if len(source) == 0 { + return nil, ErrInvalidSnapshotWatcher + } + result := make([]string, len(source)) + seen := make(map[string]struct{}, len(source)) + for index, upstream := range source { + if !workerruntime.ValidIdentifier(upstream) { + return nil, ErrInvalidSnapshotWatcher + } + if _, duplicate := seen[upstream]; duplicate { + return nil, ErrInvalidSnapshotWatcher + } + seen[upstream] = struct{}{} + result[index] = upstream + } + return result, nil +} + +func wireRoutingStrategy(source *controlplanev1.RoutingStrategy, upstreams []string) (routingDomain.Strategy, error) { + if source == nil { + return routingDomain.Strategy{}, ErrInvalidSnapshotWatcher + } + strategy := routingDomain.Strategy{CurrentUpstream: source.GetCurrentUpstream()} + switch source.GetType() { + case controlplanev1.StrategyType_STRATEGY_TYPE_SEQUENTIAL: + strategy.Type = routingDomain.StrategySequential + if !containsUpstream(upstreams, strategy.CurrentUpstream) { + return routingDomain.Strategy{}, ErrInvalidSnapshotWatcher + } + case controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM: + strategy.Type = routingDomain.StrategyRandom + case controlplanev1.StrategyType_STRATEGY_TYPE_ROUND_ROBIN: + strategy.Type = routingDomain.StrategyRoundRobin + case controlplanev1.StrategyType_STRATEGY_TYPE_WEIGHTED: + strategy.Type = routingDomain.StrategyWeighted + strategy.Weights = make(map[string]uint32, len(upstreams)) + for _, upstream := range upstreams { + weight, exists := source.GetWeights()[upstream] + if !exists || weight == 0 { + return routingDomain.Strategy{}, ErrInvalidSnapshotWatcher + } + strategy.Weights[upstream] = weight + } + for upstream := range source.GetWeights() { + if !containsUpstream(upstreams, upstream) { + return routingDomain.Strategy{}, ErrInvalidSnapshotWatcher + } + } + case controlplanev1.StrategyType_STRATEGY_TYPE_LEAST_CONNECTIONS: + strategy.Type = routingDomain.StrategyLeastConnections + default: + return routingDomain.Strategy{}, ErrInvalidSnapshotWatcher + } + return strategy, nil +} + +func wireUnavailableAction(action controlplanev1.UnavailableAction) (routingDomain.OnUnavailableAction, error) { + switch action { + case controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_REJECT: + return routingDomain.OnUnavailableReject, nil + case controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT: + return routingDomain.OnUnavailableWait, nil + case controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_DIRECT: + return routingDomain.OnUnavailableDirect, nil + default: + return "", fmt.Errorf("%w: unsupported unavailable action", ErrInvalidSnapshotWatcher) + } +} + +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 +} + func wireProxies(source []*controlplanev1.OwnedProxy) ([]proxyDomain.Proxy, error) { proxies := make([]proxyDomain.Proxy, len(source)) for index, item := range source { diff --git a/internal/gateway/controlplane/watcher_test.go b/internal/gateway/controlplane/watcher_test.go index 6d8d0e4..b11aed0 100644 --- a/internal/gateway/controlplane/watcher_test.go +++ b/internal/gateway/controlplane/watcher_test.go @@ -9,6 +9,7 @@ import ( controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/controlplane/snapshotwire" + "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/snapshot" "google.golang.org/protobuf/types/known/emptypb" @@ -23,6 +24,11 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T) Id: "proxy-a", Upstream: "upstream-a", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, Host: "192.0.2.10", Port: 8080, MaxConcurrency: 3, ExpiresAt: timestamppb.New(time.Now().Add(time.Minute)), }}, + Routing: []*controlplanev1.RoutingRule{{ + Name: "gateway-api", Enabled: true, HostRegex: `^api\.example\.test$`, Upstreams: []string{"upstream-a"}, + Strategy: &controlplanev1.RoutingStrategy{Type: controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM}, + OnUnavailable: controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_REJECT, + }}, } setSnapshotChecksum(t, full) client := &snapshotClientStub{stream: &snapshotStreamStub{values: []*controlplanev1.SnapshotEnvelope{{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: full}}}}} @@ -38,6 +44,10 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T) !view.ValidUntil.Equal(full.GetValidUntil().AsTime()) { t.Fatalf("snapshot view = %+v", view) } + matched, ok := view.MatchRouting(routing.Request{Host: "api.example.test", Method: "GET", Path: "/"}) + if !ok || matched.Name != "gateway-api" || matched.Strategy.Type != routing.StrategyRandom { + t.Fatalf("snapshot routing = %+v, %v", matched, ok) + } if client.watch.GetSessionId() != "session-a" || client.ack.GetVersion() != 1 || !client.ack.GetApplied() || string(client.ack.GetChecksum()) != string(full.GetChecksum()) { t.Fatalf("watch=%+v ack=%+v", client.watch, client.ack) } diff --git a/internal/gateway/dispatch/dispatcher.go b/internal/gateway/dispatch/dispatcher.go index f4852ff..a109587 100644 --- a/internal/gateway/dispatch/dispatcher.go +++ b/internal/gateway/dispatch/dispatcher.go @@ -6,18 +6,22 @@ import ( "time" proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/snapshot" ) var ErrNoCandidate = errors.New("no local proxy candidate is available") type Request struct { - Now time.Time - Scheme proxyDomain.Scheme - Upstreams []string - RequiredTags map[string]string - Exclude map[string]struct{} - SafetyMargin time.Duration + RoutingName string + Strategy routing.Strategy + OnUnavailable routing.OnUnavailableAction + Now time.Time + Scheme proxyDomain.Scheme + Upstreams []string + RequiredTags map[string]string + Exclude map[string]struct{} + SafetyMargin time.Duration } type Lease struct { diff --git a/internal/gateway/server/routing.go b/internal/gateway/server/routing.go index 3c543b3..980f6eb 100644 --- a/internal/gateway/server/routing.go +++ b/internal/gateway/server/routing.go @@ -6,9 +6,11 @@ import ( "net" "net/http" "strings" + "time" "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/dispatch" + "proxy-pool/internal/gateway/snapshot" ) var ( @@ -29,6 +31,37 @@ func (router *RulesRouter) Route(request *http.Request) (dispatch.Request, error if router == nil || router.rules == nil || request == nil { return dispatch.Request{}, ErrRouteNotFound } + return routeWithMatcher(request, router.rules.Match) +} + +// SnapshotRouter always matches against the rules stored with the current +// proxy snapshot so a request cannot combine routing from one version with +// proxy ownership from another. +type SnapshotRouter struct { + store *snapshot.Store +} + +func NewSnapshotRouter(store *snapshot.Store) *SnapshotRouter { + return &SnapshotRouter{store: store} +} + +func (router *SnapshotRouter) Route(request *http.Request) (dispatch.Request, error) { + if router == nil || router.store == nil || request == nil { + return dispatch.Request{}, ErrRouteNotFound + } + view := router.store.Current() + if view == nil || (!view.ValidUntil.IsZero() && !view.ValidUntil.After(time.Now().UTC())) { + return dispatch.Request{}, ErrRouteNotFound + } + return routeWithMatcher(request, view.MatchRouting) +} + +type routeMatcher func(routing.Request) (routing.Rule, bool) + +func routeWithMatcher(request *http.Request, match routeMatcher) (dispatch.Request, error) { + if request == nil || match == nil { + return dispatch.Request{}, ErrRouteNotFound + } host := request.Host if request.URL != nil && request.URL.Hostname() != "" { host = request.URL.Hostname() @@ -43,7 +76,7 @@ func (router *RulesRouter) Route(request *http.Request) (dispatch.Request, error for name := range request.Header { headers[name] = request.Header.Get(name) } - matched, ok := router.rules.Match(routing.Request{ + matched, ok := match(routing.Request{ Host: strings.ToLower(strings.TrimSuffix(host, ".")), Method: request.Method, Path: path, @@ -54,7 +87,10 @@ func (router *RulesRouter) Route(request *http.Request) (dispatch.Request, error } switch matched.Action { case routing.ActionProxy: - return dispatch.Request{Upstreams: append([]string(nil), matched.Upstreams...)}, nil + return dispatch.Request{ + RoutingName: matched.Name, Upstreams: append([]string(nil), matched.Upstreams...), + Strategy: matched.Strategy, OnUnavailable: matched.OnUnavailable, + }, nil case routing.ActionReject: return dispatch.Request{}, fmt.Errorf("%w: %s", ErrRouteRejected, matched.Name) case routing.ActionDirect: diff --git a/internal/gateway/server/routing_test.go b/internal/gateway/server/routing_test.go index 74e6727..759ec5b 100644 --- a/internal/gateway/server/routing_test.go +++ b/internal/gateway/server/routing_test.go @@ -6,8 +6,10 @@ import ( "net/http/httptest" "reflect" "testing" + "time" "proxy-pool/internal/domain/routing" + "proxy-pool/internal/gateway/snapshot" ) func TestRulesRouterReturnsMatchedUpstreams(t *testing.T) { @@ -54,3 +56,31 @@ func TestRulesRouterRejectsExplicitRejectAndMissingRoute(t *testing.T) { t.Fatalf("missing Route() error = %v", err) } } + +func TestSnapshotRouterMatchesRulesFromCurrentSnapshot(t *testing.T) { + store := snapshot.NewStore("cluster-a", "worker-a") + envelope := snapshot.Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + ValidUntil: time.Now().Add(time.Minute), + Routing: []routing.Rule{{ + Name: "gateway-api", Match: routing.Match{HostRegex: `^api\.example\.test$`}, + Upstreams: []string{"provider-a"}, Action: routing.ActionProxy, + Strategy: routing.Strategy{Type: routing.StrategyWeighted, Weights: map[string]uint32{"provider-a": 3}}, + OnUnavailable: routing.OnUnavailableReject, + }}, + } + envelope.Checksum = snapshot.ChecksumWithRouting(envelope.Proxies, envelope.Routing) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply(): %v", err) + } + router := NewSnapshotRouter(store) + + result, err := router.Route(httptest.NewRequest(http.MethodGet, "http://api.example.test/items", nil)) + if err != nil { + t.Fatalf("Route(): %v", err) + } + if result.RoutingName != "gateway-api" || result.Strategy.Type != routing.StrategyWeighted || + !reflect.DeepEqual(result.Strategy.Weights, map[string]uint32{"provider-a": 3}) { + t.Fatalf("Route() = %+v", result) + } +} diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index 33aed6d..608ddc8 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -13,6 +13,7 @@ import ( "time" proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/domain/routing" "proxy-pool/internal/domain/workerruntime" ) @@ -100,6 +101,7 @@ type Envelope struct { Checksum string ValidUntil time.Time Proxies []proxyDomain.Proxy + Routing []routing.Rule } type Entry struct { @@ -115,6 +117,7 @@ type View struct { Checksum string ValidUntil time.Time Entries []Entry + routing *routing.RuleSet all []int byScheme map[proxyDomain.Scheme][]int @@ -226,9 +229,13 @@ func (s *Store) Apply(envelope Envelope) error { if !envelope.Full || envelope.Epoch == 0 || envelope.Version == 0 { return ErrResyncRequired } - if envelope.Checksum != Checksum(envelope.Proxies) { + if envelope.Checksum != ChecksumWithRouting(envelope.Proxies, envelope.Routing) { return ErrChecksumMismatch } + routes, err := routing.Compile(envelope.Routing) + if err != nil { + return fmt.Errorf("compile snapshot routing: %w", err) + } s.mu.Lock() defer s.mu.Unlock() @@ -315,12 +322,22 @@ func (s *Store) Apply(envelope Envelope) error { Checksum: envelope.Checksum, ValidUntil: envelope.ValidUntil.UTC(), Entries: entries, + routing: routes, } next.buildIndexes() s.current.Store(next) return nil } +// MatchRouting matches a request against the immutable rules published with +// this proxy view. A View is the atomic consistency boundary for both sets. +func (v *View) MatchRouting(request routing.Request) (routing.Rule, bool) { + if v == nil || v.routing == nil { + return routing.Rule{}, false + } + return v.routing.Match(request) +} + func (s *Store) reclaimableRuntimes(nextProxyIDs map[string]struct{}) map[string]*runtimeRegistration { result := make(map[string]*runtimeRegistration) for proxyID, registration := range s.runtimes { @@ -382,8 +399,18 @@ func (s Selection) EntryAt(index int) (Entry, bool) { } func Checksum(proxies []proxyDomain.Proxy) string { - canonical := cloneAndSort(proxies) - encoded, err := json.Marshal(canonical) + return ChecksumWithRouting(proxies, nil) +} + +func ChecksumWithRouting(proxies []proxyDomain.Proxy, rules []routing.Rule) string { + payload := struct { + Proxies []proxyDomain.Proxy `json:"proxies"` + Routing []routing.Rule `json:"routing"` + }{ + Proxies: cloneAndSort(proxies), + Routing: cloneRoutingRules(rules), + } + encoded, err := json.Marshal(payload) if err != nil { panic(fmt.Sprintf("encode snapshot checksum: %v", err)) } @@ -391,6 +418,28 @@ func Checksum(proxies []proxyDomain.Proxy) string { return hex.EncodeToString(digest[:]) } +func cloneRoutingRules(source []routing.Rule) []routing.Rule { + result := make([]routing.Rule, len(source)) + for index, rule := range source { + result[index] = rule + result[index].Match.Methods = append([]string(nil), rule.Match.Methods...) + result[index].Upstreams = append([]string(nil), rule.Upstreams...) + if rule.Match.Headers != nil { + result[index].Match.Headers = make(map[string]string, len(rule.Match.Headers)) + for name, value := range rule.Match.Headers { + result[index].Match.Headers[name] = value + } + } + if rule.Strategy.Weights != nil { + result[index].Strategy.Weights = make(map[string]uint32, len(rule.Strategy.Weights)) + for upstream, weight := range rule.Strategy.Weights { + result[index].Strategy.Weights[upstream] = weight + } + } + } + return result +} + func cloneAndSort(source []proxyDomain.Proxy) []proxyDomain.Proxy { cloned := make([]proxyDomain.Proxy, len(source)) for index, descriptor := range source { diff --git a/internal/gateway/snapshot/store_test.go b/internal/gateway/snapshot/store_test.go index 7c5b789..b44239b 100644 --- a/internal/gateway/snapshot/store_test.go +++ b/internal/gateway/snapshot/store_test.go @@ -8,6 +8,7 @@ import ( "time" proxyDomain "proxy-pool/internal/domain/proxy" + "proxy-pool/internal/domain/routing" "proxy-pool/internal/domain/workerruntime" ) @@ -83,6 +84,48 @@ func TestStoreRejectsWrongWorkerVersionGapAndChecksum(t *testing.T) { } } +func TestStorePublishesRoutingWithTheSameSnapshotVersion(t *testing.T) { + store := NewStore("cluster-a", "worker-a") + routes := []routing.Rule{{ + Name: "gateway-api", + Match: routing.Match{HostRegex: `^api\.example\.test$`}, + Upstreams: []string{"provider-a"}, + Action: routing.ActionProxy, + Strategy: routing.Strategy{Type: routing.StrategyRandom}, + OnUnavailable: routing.OnUnavailableReject, + }} + envelope := Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + Routing: routes, + } + envelope.Checksum = ChecksumWithRouting(envelope.Proxies, envelope.Routing) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply(): %v", err) + } + + view := store.Current() + matched, ok := view.MatchRouting(routing.Request{Host: "api.example.test", Method: "GET", Path: "/"}) + if !ok || matched.Name != "gateway-api" || matched.Strategy.Type != routing.StrategyRandom || + !reflect.DeepEqual(matched.Upstreams, []string{"provider-a"}) { + t.Fatalf("MatchRouting() = %+v, %v", matched, ok) + } + + changedRouting := envelope + changedRouting.Version = 2 + changedRouting.Routing = []routing.Rule{{ + Name: "other", Match: routing.Match{HostRegex: `^other\.example\.test$`}, + Upstreams: []string{"provider-b"}, Action: routing.ActionProxy, + Strategy: routing.Strategy{Type: routing.StrategyRoundRobin}, OnUnavailable: routing.OnUnavailableReject, + }} + changedRouting.Checksum = Checksum(envelope.Proxies) + if err := store.Apply(changedRouting); !errors.Is(err, ErrChecksumMismatch) { + t.Fatalf("Apply(changed routing with old checksum) error = %v, want ErrChecksumMismatch", err) + } + if current := store.Current(); current.Version != 1 { + t.Fatalf("Current().Version = %d, want unchanged version 1", current.Version) + } +} + func TestViewSelectFiltersBySchemeUpstreamTagAndExclude(t *testing.T) { now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) expiresSoon := now.Add(5 * time.Second)