From 013defbd35d4bcf4beae4e8d5ff7ef6326291439 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 16:34:01 +0800 Subject: [PATCH] feat: assemble gateway process runtime --- README.md | 37 +- cmd/proxy-gateway/main.go | 97 ++++ cmd/proxy-gateway/main_test.go | 66 +++ docs/api/control-plane.md | 22 +- docs/configuration/reference.md | 18 +- docs/design/project-structure.md | 6 +- docs/development/implementation-plan.md | 6 +- docs/requirements/completion-audit.md | 16 +- docs/requirements/traceability.md | 6 +- internal/config/config.go | 9 + internal/config/config_test.go | 8 + internal/config/validate.go | 14 + internal/gateway/bootstrap/bootstrap.go | 452 +++++++++++++++++++ internal/gateway/bootstrap/bootstrap_test.go | 296 ++++++++++++ 14 files changed, 1025 insertions(+), 28 deletions(-) create mode 100644 cmd/proxy-gateway/main.go create mode 100644 cmd/proxy-gateway/main_test.go create mode 100644 internal/gateway/bootstrap/bootstrap.go create mode 100644 internal/gateway/bootstrap/bootstrap_test.go diff --git a/README.md b/README.md index 5480952..ea5dd32 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 ## 使用方式 - **Gateway**：调用方连接平台，由平台选择上游代理并转发 HTTP 或 HTTPS - CONNECT。当前已有传输、调度和保护链组件，命令进程与控制面快照客户端待装配。 + CONNECT。`proxy-gateway` 已装配本地监听、指标探针和控制面 Register/Watch/ACK/ + Runtime 会话；带凭据 Proxy 分发仍待闭环。 - **Distribution**：调用方按条件提取真实代理；成功提取即独占消费，不支持归还、 续租或状态查询。 - **Admin**：运维人员查询状态、启停 Upstream、切换 Routing，并触发严格配置 @@ -47,7 +48,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 Routing 快照。Gateway 会将 Routing 与 Proxy 原子编译为同一内存 View，并只按当前未过期 View 匹配请求，并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待， - 以及仍经过目标地址策略的 direct；`proxy-gateway` 命令与凭据分发仍待装配。 + 以及仍经过目标地址策略的 direct；`proxy-gateway` 通过独立控制面拨号地址维护 + Session，并仅在持有未过期 Snapshot 时 Ready；凭据分发仍待装配。 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化。 @@ -55,7 +57,7 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 ```mermaid flowchart LR - Client[调用方] -->|HTTP / CONNECT| Gateway[Gateway
进程待装配] + Client[调用方] -->|HTTP / CONNECT| Gateway[proxy-gateway] Client -->|独占提取| Distribution[Distribution] Operator[运维人员] -->|管理操作| Admin[Admin] subgraph CP[proxy-controller 已运行] @@ -66,7 +68,7 @@ flowchart LR Controller --> Redis[(Redis)] Controller --> PostgreSQL[(PostgreSQL)] Checker[Checker
健康链待闭环] -. Observation .-> Controller - Controller -. Snapshot 链待闭环 .-> Gateway + Controller -. gRPC Snapshot .-> Gateway ``` - **PostgreSQL** 只保存管理面状态，不保存 Proxy 明细或逐次提取记录。 @@ -76,7 +78,7 @@ flowchart LR ## 当前完成度 -截至 **2026-07-31**，实施计划检查项为 **53 / 74（71.6%）**。详情见 +截至 **2026-07-31**，实施计划检查项为 **54 / 74（73.0%）**。详情见 [实施计划](docs/development/implementation-plan.md)和 [交付完成度审计](docs/requirements/completion-audit.md)。 @@ -84,9 +86,9 @@ flowchart LR 提取与限流、Controller 的 Admin/Distribution/Metrics 监听，以及 PostgreSQL 管理状态；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和 Redis 会话栅栏。 -- **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership - 与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。 -- **待完成**：带凭据 Proxy 分发、Gateway 进程装配、Outcome 上报、Checker 调度与健康状态链， +- **部分完成**：Gateway 的带凭据 Proxy 分发与 Outcome 上报、Checker 调度与健康状态链、 + Docker Compose/Kubernetes 运行时 mTLS Overlay，以及 protobuf 契约。 +- **待完成**：带凭据 Proxy 分发、Outcome 上报、Checker 调度与健康状态链， 以及 loadgen 和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 @@ -137,7 +139,20 @@ go run ./cmd/proxy-controller -config CONFIG_FILE 改用宿主机可达的存储地址。 本地配置中的 `.invalid` Provider URL 是故障演示占位，不会提供真实代理。 -当前仓库没有 Gateway、Checker 或 loadgen 命令，因此不提供对应启动命令。 +Gateway 已提供启动命令；需要先启用 Controller `controlPlane` 并配置匹配的 mTLS +证书（回环 fixture 可使用明文），再提供独立的拨号地址和 Worker 身份： + +```powershell +go run ./cmd/proxy-gateway -config CONFIG_FILE ` + -control-plane CONTROLLER_HOST:8443 ` + -cluster-id CLUSTER_ID -worker-id WORKER_ID ` + -instance-id INSTANCE_ID -zone ZONE +``` + +以上参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、`PROXY_POOL_CLUSTER_ID`、 +`PROXY_POOL_WORKER_ID`、`PROXY_POOL_INSTANCE_ID` 与 `PROXY_POOL_ZONE` 提供。 +Gateway 的 `/livez`、`/readyz`、`/metrics` 使用配置中的 `metrics.listen`；无有效 +Snapshot 时 `/readyz` 返回 `503`。Checker 与 loadgen 命令尚未实现。 ## 关键配置与入口 @@ -170,8 +185,8 @@ go run ./cmd/proxy-controller -config CONFIG_FILE ownership 索引，以及 Gateway 快照客户端。 - **P0 - Checker 健康链**：Checker 调度、实际探测、Observation reducer，以及 `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链。 -- **P1 - Gateway 与 Routing**：`proxy-gateway` 命令、五种 Routing 策略、 - `onUnavailable`、动态容量调整和 Drain 闭环。 +- **P1 - Gateway 与 Routing**：Gateway 进程、五种 Routing 策略与 + `onUnavailable` 已接入；带凭据 Proxy 分发、动态容量调整和 Drain 闭环待完成。 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑， 以及故障转移和恢复演练。 - **P2 - 容量证明**：`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测，以及 diff --git a/cmd/proxy-gateway/main.go b/cmd/proxy-gateway/main.go new file mode 100644 index 0000000..6735616 --- /dev/null +++ b/cmd/proxy-gateway/main.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "os" + "os/signal" + "strings" + "syscall" + + "proxy-pool/internal/config" + "proxy-pool/internal/gateway/bootstrap" +) + +const ( + configEnvironment = "PROXY_POOL_CONFIG" + controlPlaneAddressEnvironment = "PROXY_POOL_CONTROL_PLANE_ADDRESS" + clusterIDEnvironment = "PROXY_POOL_CLUSTER_ID" + workerIDEnvironment = "PROXY_POOL_WORKER_ID" + instanceIDEnvironment = "PROXY_POOL_INSTANCE_ID" + zoneEnvironment = "PROXY_POOL_ZONE" +) + +type environmentLookup func(string) string +type gatewayRun func(context.Context, bootstrap.Options) error + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + os.Exit(execute(ctx, os.Args[1:], os.Getenv, bootstrap.Run, os.Stderr)) +} + +func execute( + ctx context.Context, + args []string, + getenv environmentLookup, + run gatewayRun, + stderr io.Writer, +) int { + flags := flag.NewFlagSet("proxy-gateway", flag.ContinueOnError) + flags.SetOutput(stderr) + configPath := flags.String("config", "", "configuration file path") + controlPlaneAddress := flags.String("control-plane", "", "remote Controller control-plane address") + clusterID := flags.String("cluster-id", "", "cluster identifier") + workerID := flags.String("worker-id", "", "unique Worker identifier") + instanceID := flags.String("instance-id", "", "unique process instance identifier") + zone := flags.String("zone", "", "availability zone identifier") + if err := flags.Parse(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + return 0 + } + return 2 + } + if flags.NArg() != 0 { + _, _ = fmt.Fprintln(stderr, "proxy-gateway: unexpected positional arguments") + return 2 + } + if getenv != nil { + setIfEmpty(configPath, getenv(configEnvironment)) + setIfEmpty(controlPlaneAddress, getenv(controlPlaneAddressEnvironment)) + setIfEmpty(clusterID, getenv(clusterIDEnvironment)) + setIfEmpty(workerID, getenv(workerIDEnvironment)) + setIfEmpty(instanceID, getenv(instanceIDEnvironment)) + setIfEmpty(zone, getenv(zoneEnvironment)) + } + if ctx == nil || run == nil || !validValue(*configPath) || !validValue(*controlPlaneAddress) || + !validValue(*clusterID) || !validValue(*workerID) || !validValue(*instanceID) || !validValue(*zone) { + _, _ = fmt.Fprintf(stderr, + "proxy-gateway: -config, -control-plane, -cluster-id, -worker-id, -instance-id and -zone are required; "+ + "environment fallbacks: %s, %s, %s, %s, %s, %s\n", + configEnvironment, controlPlaneAddressEnvironment, clusterIDEnvironment, workerIDEnvironment, instanceIDEnvironment, zoneEnvironment, + ) + return 2 + } + err := run(ctx, bootstrap.Options{ + ConfigPath: *configPath, Resolver: config.OSResolver{}, ControlPlaneAddress: *controlPlaneAddress, + ClusterID: *clusterID, WorkerID: *workerID, InstanceID: *instanceID, Zone: *zone, + }) + if err == nil || (errors.Is(err, context.Canceled) && ctx.Err() != nil) { + return 0 + } + _, _ = fmt.Fprintf(stderr, "proxy-gateway: %v\n", err) + return 1 +} + +func setIfEmpty(target *string, value string) { + if target != nil && *target == "" { + *target = value + } +} + +func validValue(value string) bool { + return value != "" && strings.TrimSpace(value) == value +} diff --git a/cmd/proxy-gateway/main_test.go b/cmd/proxy-gateway/main_test.go new file mode 100644 index 0000000..d6ba26b --- /dev/null +++ b/cmd/proxy-gateway/main_test.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "io" + "strings" + "testing" + + "proxy-pool/internal/gateway/bootstrap" +) + +func TestExecutePassesExplicitGatewayIdentity(t *testing.T) { + t.Parallel() + + var captured bootstrap.Options + status := execute(context.Background(), []string{ + "-config", "gateway.yaml", + "-control-plane", "127.0.0.1:8443", + "-cluster-id", "cluster-a", + "-worker-id", "worker-a", + "-instance-id", "instance-a", + "-zone", "zone-a", + }, func(string) string { return "" }, func(_ context.Context, options bootstrap.Options) error { + captured = options + return nil + }, io.Discard) + if status != 0 { + t.Fatalf("execute() status = %d, want 0", status) + } + if captured.ConfigPath != "gateway.yaml" || captured.ControlPlaneAddress != "127.0.0.1:8443" || + captured.ClusterID != "cluster-a" || captured.WorkerID != "worker-a" || + captured.InstanceID != "instance-a" || captured.Zone != "zone-a" { + t.Fatalf("bootstrap options = %+v", captured) + } +} + +func TestExecuteReadsGatewayOptionsFromEnvironment(t *testing.T) { + t.Parallel() + + values := map[string]string{ + configEnvironment: "gateway.yaml", + controlPlaneAddressEnvironment: "127.0.0.1:8443", + clusterIDEnvironment: "cluster-a", + workerIDEnvironment: "worker-a", + instanceIDEnvironment: "instance-a", + zoneEnvironment: "zone-a", + } + var captured bootstrap.Options + status := execute(context.Background(), nil, func(name string) string { return values[name] }, func(_ context.Context, options bootstrap.Options) error { + captured = options + return nil + }, io.Discard) + if status != 0 || captured.WorkerID != "worker-a" || captured.ControlPlaneAddress != "127.0.0.1:8443" { + t.Fatalf("execute() = (%d, %+v)", status, captured) + } +} + +func TestExecuteRejectsIncompleteGatewayOptions(t *testing.T) { + t.Parallel() + + var output strings.Builder + status := execute(context.Background(), []string{"-config", "gateway.yaml"}, func(string) string { return "" }, nil, &output) + if status != 2 || !strings.Contains(output.String(), controlPlaneAddressEnvironment) { + t.Fatalf("execute() = (%d, %q), want usage error naming missing environment", status, output.String()) + } +} diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 429b642..f9e862e 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -33,8 +33,10 @@ Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有 Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本地 View，动态 Router 只匹配 当前未过期 View；派发器已按五种策略从该 View 选择上游，且在 Proxy 容量耗尽时只在该 View 的其余候选中回退。`wait_timeout` 随 `on_unavailable=WAIT` 下发并在 Gateway 作为有界 -本地容量等待使用；`DIRECT` 仍先经过 TargetPolicy 再建立 HTTP/CONNECT 直连。增量、Gateway -进程装配、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`； +本地容量等待使用；`DIRECT` 仍先经过 TargetPolicy 再建立 HTTP/CONNECT 直连。`proxy-gateway` +已装配 Register/Watch/ACK/Runtime 会话、HTTP 代理监听和 Snapshot 就绪探针；控制面中断时 +保持进程运行并以有界退避重连，未取得有效 Snapshot 的 Worker 不会 Ready。增量、带凭据 +Proxy 分发、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`； `100,000 QPS` 仍是未验证的设计目标。 `WatchSnapshots` 建立时校验当前 session；每次签发快照引用时也把 `session_id` @@ -163,3 +165,19 @@ Routing 决定 AVAILABLE、SUSPECT 或 UNHEALTHY，并更新 Redis 活动池， 集群环境使用 mTLS，证书身份绑定 Worker/Checker 类型和环境。服务端校验 消息中的逻辑 ID 与证书授权一致，设置单消息大小、流持续时间、并发 Stream 和上报批次上限。`secret_ref` 是受控引用，不在 Proto 中传播真实密码。 + +## 9. Gateway 启动参数 + +Gateway 不复用 `controlPlane.listen` 作为客户端地址。`listen` 是 Controller 的 +服务端绑定地址；每个 Gateway 必须显式提供以下独立参数或同名环境变量： + +- `-control-plane` / `PROXY_POOL_CONTROL_PLANE_ADDRESS`：Controller 的可拨号地址。 +- `-cluster-id` / `PROXY_POOL_CLUSTER_ID`：快照所属集群。 +- `-worker-id` / `PROXY_POOL_WORKER_ID`：唯一逻辑 Worker。 +- `-instance-id` / `PROXY_POOL_INSTANCE_ID`：唯一进程实例。 +- `-zone` / `PROXY_POOL_ZONE`：实例可用区。 + +当 `controlPlane.tls.mode=mtls` 时，Gateway 使用 `controlPlane.gatewayTLS` 中独立的 +客户端证书、私钥和 Controller CA 发起 TLS 1.3 连接；证书必须符合 Controller 的 +SPIFFE Worker 身份校验。 +`disabled` 仅接受回环控制面地址，供本地 fixture 使用。 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 348d363..eb4e2d2 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -21,8 +21,9 @@ go run ./cmd/proxy-controller -config CONFIG_FILE 所有 Controller 副本必须一致，且该密钥不得放入 YAML 或 PostgreSQL。该入口已装配 PostgreSQL 管理面迁移、Redis 活动池、Distribution/Admin 独立监听与优雅停机； Controller Metrics 独立监听、`/livez`、`/readyz` 和基础 Prometheus 运行时指标； -Provider 自动补池、分布式配额、动态重载和 Admin 低基数统计已装配。完整 Gateway、 -Checker、Worker 控制面与代表性负载验证仍在后续实施范围。 +Provider 自动补池、分布式配额、动态重载和 Admin 低基数统计已装配。Gateway 进程、 +Worker 控制面会话和 Snapshot 就绪探针已装配；Checker、带凭据 Proxy 分发与代表性 +负载验证仍在后续实施范围。 所有时间值使用 Go duration，例如 `500ms`、`30s`、`5m`。示例中的 `${TOKEN}`、`${PASSWORD}`、`${POSTGRES_URL}` 等由加载器从同名环境变量 @@ -166,6 +167,10 @@ controlPlane: clientCAFile: /run/secrets/worker-ca.pem trustDomain: proxy.example environment: production + gatewayTLS: + certFile: /run/secrets/gateway-cert.pem + keyFile: /run/secrets/gateway-key.pem + serverCAFile: /run/secrets/controller-ca.pem ``` - `protocolVersion` 当前固定为 `1`。 @@ -175,6 +180,15 @@ controlPlane: - `tls.mode` 只能为 `disabled` 或 `mtls`。`disabled` 只允许回环监听；`mtls` 必须 同时配置证书、私钥、客户端 CA、全小写 DNS `trustDomain` 和单 URI 路径段 `environment`。 +- `gatewayTLS` 是 Gateway 的客户端证书、私钥和 Controller CA，与 `tls` 的服务端 + 证书和 Worker CA 分离。三项可以同时省略（未运行 Gateway），配置任一项时必须完整提供。 + +Gateway 连接 Controller 时使用独立启动参数而非 `controlPlane.listen`。至少设置 +`PROXY_POOL_CONTROL_PLANE_ADDRESS`、`PROXY_POOL_CLUSTER_ID`、 +`PROXY_POOL_WORKER_ID`、`PROXY_POOL_INSTANCE_ID` 和 `PROXY_POOL_ZONE`，详见 +[控制面协议](../api/control-plane.md#9-gateway-启动参数)。基础 Compose/Kubernetes +模板保持 `controlPlane.enabled: false`，环境 Overlay 挂载 mTLS 证书并启用后才可启动 +Gateway。 ## 5. Gateway diff --git a/docs/design/project-structure.md b/docs/design/project-structure.md index ea4fbdd..c58a0c8 100644 --- a/docs/design/project-structure.md +++ b/docs/design/project-structure.md @@ -12,7 +12,7 @@ proxy-pool/ ├── internal/ │ ├── config/ # 严格配置解析和校验 │ ├── domain/ # 无传输、无存储依赖的领域模型 -│ ├── gateway/ # snapshot、dispatch、server、transport +│ ├── gateway/ # bootstrap、snapshot、dispatch、server、transport │ ├── controller/ # provider、pool、extraction、operations、runtime、bootstrap │ ├── adapters/ # PostgreSQL、Redis、Provider API、内存适配 │ └── platform/ # HTTP、安全、日志、指标、停机和进程装配 @@ -34,6 +34,10 @@ proxy-pool/ 选择、session 与重试资格等热路径决策；`transport` 独占连接池、上游握手和 隧道生命周期。任何包都不得从热路径反向调用 Controller 存储。 +`gateway/bootstrap` 只在进程启动时加载配置、建立 gRPC 控制面 Session、组装 HTTP +代理与 Metrics 监听；请求热路径只读取本地 Snapshot。`/readyz` 要求当前 Snapshot +尚未到期，控制面断开期间以有界退避重连而不访问 Redis/PostgreSQL。 + ### proxy-controller Controller 是首版模块化单体。Provider、Pool、Routing 和 Extraction 共享 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 2030837..5b7e9b2 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -212,7 +212,8 @@ Prometheus 运行时指标，三监听器隔离已通过测试；业务指标仍 WorkerControlPlane 现已接入 Controller 生命周期：Register、ACK 和 Runtime 报告均经 Redis 服务端 TTL 的 session/issued-snapshot/ACK 栅栏校验；mTLS SPIFFE 身份、消息/流限制和有界停机已实现。`WatchSnapshots` 会发送当前 epoch 的基础完整 -Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话协调组件。 +Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话协调组件与 +`proxy-gateway` 进程装配。 按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本，并可构建无凭据引用的 已归属 Proxy payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成， 重注册会清除旧引用，避免迟到 Stream 覆盖新 session。Worker 服务端会在最近完整 @@ -221,7 +222,8 @@ Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已 payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 checksum；Gateway 已将其与 Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。派发器的五种上游选择已 接入该 View，并在容量耗尽时在同版本候选中回退；`onUnavailable` 的 reject、wait 与 direct -已接入 Gateway，凭据分发、Gateway 命令与 Outcome 上报仍未实现。 +已接入 Gateway；`proxy-gateway` 已装配本地 HTTP/Metrics 监听、快照就绪探针和 +控制面重连，凭据分发与 Outcome 上报仍未实现。 已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 2bd444f..317d37d 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -37,7 +37,9 @@ - `OPS-001`：完整 Snapshot 目标、epoch/version、校验和验证及原子替换。 - `CAP-001 / GW 热路径边界`：本地 Dispatch 条件过滤与原子容量预留。 - `GW-*`：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、超时、重试、保护链与 - 优雅停机 Handler 已实现并通过定向测试。 + 优雅停机 Handler 已实现并通过定向测试。`proxy-gateway` 已装配本地 HTTP/Metrics + 监听、快照就绪探针、显式控制面拨号和 Register/Watch/ACK/Runtime 会话；本地 gRPC + 集成测试验证收到有效 Snapshot 后才 Ready，并可完成 direct 路由。 - `PROVIDER-*`：Provider HTTP Client、严格响应上限、模板解析安全边界、凭据 引用 Store 与 Reconciler Adapter 已实现。 - `DIST/Admin HTTP`：严格 JSON、Request ID、Problem 响应及 Distribution/Admin @@ -88,10 +90,10 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt 以下已有设计、接口或部署位置，但尚无端到端生产实现： -1. `cmd/proxy-gateway/checker/loadgen` 进程装配；`proxy-controller` 已完成 - Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，但 Provider 和业务 - 指标链未闭环。 -2. Gateway 进程装配、生产连接池调优与代表性流量压测。 +1. `cmd/proxy-checker/loadgen` 进程装配；`proxy-controller` 已完成 + Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成 + HTTP/Metrics 与控制面 Session 装配，但 Provider 和业务指标链未闭环。 +2. Gateway 的带凭据 Proxy 分发、生产连接池调优与代表性流量压测。 3. Provider 分布式 singleflight/Leader、长期凭据回收和累计额度执行器。 4. Controller 的 PostgreSQL 连接池、迁移和 pgx Adapter 启动装配已完成； 公用 bootstrap 已通过 PostgreSQL 18 + Redis 8.2 双存储集成，Controller @@ -100,8 +102,8 @@ 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 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成；凭据 - 分发与 Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。 + 客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成； + 凭据分发与 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 3f5608a..bc38337 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -7,9 +7,9 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| 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-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC，并发布无凭据 Proxy/Gateway Routing 完整快照；Gateway 已将同版本 Routing 编译为动态匹配 View，并由独立进程维护控制面会话。Checker/Loadgen 构建产物与带凭据 Proxy 分发待实现 | +| ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；代表性性能剖析待完成 | +| ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 | | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | | ARCH-005 | 100k QPS 峰值使用多 Worker 集群 | 当前会话 | 未验证设计目标；待代表性集群负载报告 | diff --git a/internal/config/config.go b/internal/config/config.go index ff8c26a..2e4d57b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -139,6 +139,7 @@ type ControlPlane struct { MaxRuntimeCounters int `yaml:"maxRuntimeCounters"` MaxConcurrentStreams uint32 `yaml:"maxConcurrentStreams"` TLS ControlPlaneTLS `yaml:"tls"` + GatewayTLS GatewayTLS `yaml:"gatewayTLS"` } type ControlPlaneTLS struct { @@ -150,6 +151,14 @@ type ControlPlaneTLS struct { Environment string `yaml:"environment"` } +// GatewayTLS holds client-only mTLS material. It remains separate from the +// Controller's server certificate and CA configuration. +type GatewayTLS struct { + CertFile string `yaml:"certFile"` + KeyFile string `yaml:"keyFile"` + ServerCAFile string `yaml:"serverCAFile"` +} + type Storage struct { PostgresURL string `yaml:"postgresURL"` RedisURL string `yaml:"redisURL"` diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 2a2b6a1..def7cdc 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -454,6 +454,14 @@ func TestValidateControlPlane(t *testing.T) { }, want: "environment", }, + { + name: "partial gateway tls", + mutate: func(cfg *Config) { + cfg.ControlPlane = validMTLSControlPlane() + cfg.ControlPlane.GatewayTLS.CertFile = "/run/secrets/gateway-cert.pem" + }, + want: "gatewayTLS", + }, } for _, test := range tests { diff --git a/internal/config/validate.go b/internal/config/validate.go index ac4b780..e0a6c30 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -110,6 +110,9 @@ func Validate(cfg *Config) error { } func validateControlPlane(item ControlPlane) error { + if err := validateGatewayTLS(item.GatewayTLS); err != nil { + return err + } if !item.Enabled { return nil } @@ -161,6 +164,17 @@ func validateControlPlane(item ControlPlane) error { return nil } +func validateGatewayTLS(item GatewayTLS) error { + configured := item.CertFile != "" || item.KeyFile != "" || item.ServerCAFile != "" + if !configured { + return nil + } + if item.CertFile == "" || item.KeyFile == "" || item.ServerCAFile == "" { + return fmt.Errorf("validate controlPlane gatewayTLS: certFile, keyFile and serverCAFile are required together") + } + return nil +} + func validTrustDomain(value string) bool { if len(value) == 0 || len(value) > 253 { return false diff --git a/internal/gateway/bootstrap/bootstrap.go b/internal/gateway/bootstrap/bootstrap.go new file mode 100644 index 0000000..cff1911 --- /dev/null +++ b/internal/gateway/bootstrap/bootstrap.go @@ -0,0 +1,452 @@ +// Package bootstrap assembles the proxy-gateway process from data-plane +// components without putting Controller or storage access on the request path. +package bootstrap + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "net" + "os" + "reflect" + "strconv" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/config" + "proxy-pool/internal/domain/workerruntime" + "proxy-pool/internal/gateway/controlplane" + "proxy-pool/internal/gateway/dispatch" + "proxy-pool/internal/gateway/server" + "proxy-pool/internal/gateway/snapshot" + "proxy-pool/internal/gateway/transport" + "proxy-pool/internal/platform/httpserver" + "proxy-pool/internal/platform/lifecycle" + platformMetrics "proxy-pool/internal/platform/metrics" +) + +var ( + ErrInvalidOptions = errors.New("invalid gateway bootstrap options") + ErrStartup = errors.New("gateway startup failed") + ErrNotReady = errors.New("gateway snapshot is not ready") +) + +const ( + defaultReconnectInitialDelay = time.Second + defaultReconnectMaxDelay = 30 * time.Second + defaultReconnectJitter = 20 +) + +// Options provides process-local settings. ControlPlaneAddress is deliberately +// independent from controlPlane.listen: the latter is the Controller bind +// address, while this value is the Gateway's remote dial target. +type Options struct { + ConfigPath string + Resolver config.Resolver + ControlPlaneAddress string + ClusterID string + WorkerID string + InstanceID string + Zone string + Labels map[string]string + HTTP httpserver.Options + + // The pre-bound listeners and transport are test seams. Production leaves + // them nil and derives listeners and mTLS credentials from configuration. + GatewayListener net.Listener + MetricsListener net.Listener + GRPCTransport credentials.TransportCredentials + + ReconnectInitialDelay time.Duration + ReconnectMaxDelay time.Duration + ReconnectJitter int +} + +func Run(ctx context.Context, options Options) error { + if err := validateOptions(ctx, options); err != nil { + return err + } + configuration, err := loadConfiguration(ctx, options.ConfigPath, options.Resolver) + if err != nil { + return fmt.Errorf("%w: load configuration: %w", ErrStartup, err) + } + if !configuration.Gateway.Enabled || !configuration.ControlPlane.Enabled { + return errors.Join(ErrInvalidOptions, errors.New("gateway and controlPlane must be enabled")) + } + + runtime, err := newRuntime(ctx, configuration, options) + if err != nil { + return fmt.Errorf("%w: %w", ErrStartup, err) + } + defer runtime.Close() + return runtime.Run(ctx) +} + +func validateOptions(ctx context.Context, options Options) error { + if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" || + nilInterface(options.Resolver) || !workerruntime.ValidIdentifier(options.ClusterID) || !workerruntime.ValidIdentifier(options.WorkerID) || + !workerruntime.ValidIdentifier(options.InstanceID) || !workerruntime.ValidIdentifier(options.Zone) || !validDialAddress(options.ControlPlaneAddress) { + return ErrInvalidOptions + } + if options.GatewayListener == nil && options.MetricsListener != nil { + return ErrInvalidOptions + } + if options.ReconnectInitialDelay < 0 || options.ReconnectMaxDelay < 0 || options.ReconnectJitter < 0 || options.ReconnectJitter > 100 { + return ErrInvalidOptions + } + return nil +} + +func loadConfiguration(ctx context.Context, path string, resolver config.Resolver) (*config.Config, error) { + if ctx == nil || strings.TrimSpace(path) != path || path == "" || nilInterface(resolver) { + return nil, ErrInvalidOptions + } + if err := ctx.Err(); err != nil { + return nil, err + } + content, err := resolver.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read configuration %q: %w", path, err) + } + if err := ctx.Err(); err != nil { + return nil, err + } + configuration, err := config.LoadResolved(bytes.NewReader(content), resolver) + if err != nil { + return nil, fmt.Errorf("decode configuration %q: %w", path, err) + } + return configuration, nil +} + +type runtime struct { + connection *grpc.ClientConn + group *lifecycle.Group +} + +func newRuntime(ctx context.Context, configuration *config.Config, options Options) (*runtime, error) { + if ctx == nil || configuration == nil { + return nil, ErrInvalidOptions + } + store := snapshot.NewStore(options.ClusterID, options.WorkerID) + protection, err := server.BuildProtection(configuration.Gateway) + if err != nil { + return nil, fmt.Errorf("build gateway protections: %w", err) + } + targets, err := server.TargetPolicyFromListener(configuration.Gateway) + if err != nil { + return nil, fmt.Errorf("build gateway target policy: %w", err) + } + proxyTransport := transport.New(transport.Config{}, nil) + handler, err := server.New(server.ConfigFromListener(configuration.Gateway), server.Dependencies{ + Auth: protection.Auth, + Access: protection.Access, + Admission: protection.Admission, + Targets: targets, + Router: server.NewSnapshotRouter(store), + Dispatcher: dispatch.New(store), + Transport: proxyTransport, + }) + if err != nil { + proxyTransport.CloseIdleConnections() + return nil, fmt.Errorf("build gateway handler: %w", err) + } + + transportCredentials, err := controlPlaneTransport(configuration.ControlPlane, options.ControlPlaneAddress, options.GRPCTransport) + if err != nil { + proxyTransport.CloseIdleConnections() + return nil, err + } + connection, err := grpc.NewClient(options.ControlPlaneAddress, + grpc.WithTransportCredentials(transportCredentials), + grpc.WithDefaultCallOptions( + grpc.MaxCallRecvMsgSize(configuration.ControlPlane.MaxMessageBytes), + grpc.MaxCallSendMsgSize(configuration.ControlPlane.MaxMessageBytes), + ), + ) + if err != nil { + proxyTransport.CloseIdleConnections() + return nil, fmt.Errorf("dial control plane: %w", err) + } + + closeConnection := true + defer func() { + if closeConnection { + _ = connection.Close() + proxyTransport.CloseIdleConnections() + } + }() + client := controlplanev1.NewWorkerControlPlaneClient(connection) + reporter, err := controlplane.NewRuntimeReporter(generatedRuntimeClient{client: client}, store, controlplane.Options{ + WorkerID: options.WorkerID, InstanceID: options.InstanceID, Zone: options.Zone, + ProtocolVersion: configuration.ControlPlane.ProtocolVersion, Labels: options.Labels, Now: time.Now, + }) + if err != nil { + return nil, fmt.Errorf("build runtime reporter: %w", err) + } + watcher, err := controlplane.NewSnapshotWatcher(controlplane.NewGeneratedSnapshotRPCClient(client), store, + controlplane.SnapshotWatcherOptions{ClusterID: options.ClusterID, WorkerID: options.WorkerID}) + if err != nil { + return nil, fmt.Errorf("build snapshot watcher: %w", err) + } + session, err := controlplane.NewSessionRunner(reporter, watcher) + if err != nil { + return nil, fmt.Errorf("build control plane session: %w", err) + } + reconnect, err := controlplane.NewSessionSupervisor(session, reconnectOptions(options)) + if err != nil { + return nil, fmt.Errorf("build control plane reconnect supervisor: %w", err) + } + + httpRuntime, err := newHTTPRuntime(ctx, configuration, options, handler, store) + if err != nil { + return nil, err + } + group, err := lifecycle.NewGroup(httpRuntime, reconnect) + if err != nil { + return nil, err + } + closeConnection = false + return &runtime{connection: connection, group: group}, nil +} + +type generatedRuntimeClient struct { + client controlplanev1.WorkerControlPlaneClient +} + +func (client generatedRuntimeClient) RegisterWorker( + ctx context.Context, + request *controlplanev1.RegisterWorkerRequest, +) (*controlplanev1.RegisterWorkerResponse, error) { + return client.client.RegisterWorker(ctx, request) +} + +func (client generatedRuntimeClient) ReportRuntime( + ctx context.Context, + request *controlplanev1.ReportRuntimeRequest, +) (*controlplanev1.ReportRuntimeResponse, error) { + return client.client.ReportRuntime(ctx, request) +} + +func (runtime *runtime) Run(ctx context.Context) error { + if runtime == nil || runtime.group == nil || ctx == nil { + return ErrInvalidOptions + } + return runtime.group.Run(ctx) +} + +func (runtime *runtime) Close() { + if runtime != nil && runtime.connection != nil { + _ = runtime.connection.Close() + } +} + +func reconnectOptions(options Options) controlplane.ReconnectOptions { + initial := options.ReconnectInitialDelay + if initial == 0 { + initial = defaultReconnectInitialDelay + } + maximum := options.ReconnectMaxDelay + if maximum == 0 { + maximum = defaultReconnectMaxDelay + } + jitter := options.ReconnectJitter + if jitter == 0 { + jitter = defaultReconnectJitter + } + return controlplane.ReconnectOptions{InitialDelay: initial, MaxDelay: maximum, Jitter: jitter} +} + +type httpRuntime struct { + options httpserver.Options + handler *server.Handler + endpoints []httpserver.Endpoint +} + +func newHTTPRuntime( + ctx context.Context, + configuration *config.Config, + options Options, + handler *server.Handler, + store *snapshot.Store, +) (*httpRuntime, error) { + if ctx == nil || configuration == nil || handler == nil || store == nil { + return nil, ErrInvalidOptions + } + endpoints := make([]httpserver.Endpoint, 0, 2) + gatewayListener, gatewayOwned, err := resolveListener(ctx, configuration.Gateway.Listen, options.GatewayListener) + if err != nil { + return nil, fmt.Errorf("listen gateway: %w", err) + } + endpoints = append(endpoints, httpserver.Endpoint{Name: "gateway", Listener: gatewayListener, Handler: handler}) + if configuration.Metrics.Enabled { + metricsHandler, metricsErr := platformMetrics.NewHandler(platformMetrics.Dependencies{ + Gatherer: prometheus.DefaultGatherer, + Readiness: snapshotReadiness{store: store, now: time.Now}, + }) + if metricsErr != nil { + if gatewayOwned { + _ = gatewayListener.Close() + } + return nil, fmt.Errorf("build gateway metrics: %w", metricsErr) + } + metricsListener, metricsOwned, listenErr := resolveListener(ctx, configuration.Metrics.Listen, options.MetricsListener) + if listenErr != nil { + if gatewayOwned { + _ = gatewayListener.Close() + } + return nil, fmt.Errorf("listen gateway metrics: %w", listenErr) + } + _ = metricsOwned + endpoints = append(endpoints, httpserver.Endpoint{Name: "metrics", Listener: metricsListener, Handler: metricsHandler}) + } else if options.MetricsListener != nil { + if gatewayOwned { + _ = gatewayListener.Close() + } + return nil, ErrInvalidOptions + } + return &httpRuntime{options: options.HTTP, handler: handler, endpoints: endpoints}, nil +} + +func (runtime *httpRuntime) Run(ctx context.Context) error { + if runtime == nil || runtime.handler == nil || len(runtime.endpoints) == 0 || ctx == nil { + return ErrInvalidOptions + } + completed := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + shutdownContext, cancel := context.WithTimeout(context.Background(), shutdownTimeout(runtime.options)) + defer cancel() + _ = runtime.handler.Shutdown(shutdownContext) + case <-completed: + } + }() + err := httpserver.Serve(ctx, runtime.options, runtime.endpoints...) + close(completed) + return err +} + +func resolveListener(ctx context.Context, address string, listener net.Listener) (net.Listener, bool, error) { + if listener != nil { + return listener, false, nil + } + resolved, err := (&net.ListenConfig{}).Listen(ctx, "tcp", address) + if err != nil { + return nil, false, err + } + return resolved, true, nil +} + +func shutdownTimeout(options httpserver.Options) time.Duration { + if options.ShutdownTimeout > 0 { + return options.ShutdownTimeout + } + return httpserver.DefaultOptions().ShutdownTimeout +} + +type snapshotReadiness struct { + store *snapshot.Store + now func() time.Time +} + +func (readiness snapshotReadiness) Ready(ctx context.Context) error { + if ctx == nil || readiness.store == nil || readiness.now == nil { + return ErrNotReady + } + if err := ctx.Err(); err != nil { + return err + } + current := readiness.store.Current() + if current == nil || current.ValidUntil.IsZero() || !current.ValidUntil.After(readiness.now().UTC()) { + return ErrNotReady + } + return nil +} + +func controlPlaneTransport( + configuration config.ControlPlane, + address string, + override credentials.TransportCredentials, +) (credentials.TransportCredentials, error) { + if override != nil { + return override, nil + } + switch configuration.TLS.Mode { + case "disabled": + if !loopbackAddress(address) { + return nil, fmt.Errorf("%w: plaintext control plane target must be loopback", ErrInvalidOptions) + } + return insecure.NewCredentials(), nil + case "mtls": + if configuration.GatewayTLS.CertFile == "" || configuration.GatewayTLS.KeyFile == "" || configuration.GatewayTLS.ServerCAFile == "" { + return nil, fmt.Errorf("%w: controlPlane.gatewayTLS is required for mtls", ErrInvalidOptions) + } + certificate, err := tls.LoadX509KeyPair(configuration.GatewayTLS.CertFile, configuration.GatewayTLS.KeyFile) + if err != nil { + return nil, fmt.Errorf("load gateway control plane certificate: %w", err) + } + caPEM, err := os.ReadFile(configuration.GatewayTLS.ServerCAFile) + if err != nil { + return nil, fmt.Errorf("read gateway control plane CA: %w", err) + } + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(caPEM) { + return nil, errors.New("parse gateway control plane CA") + } + host, _, _ := net.SplitHostPort(address) + return credentials.NewTLS(&tls.Config{ + MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate}, RootCAs: roots, + ServerName: strings.Trim(host, "[]"), + }), nil + default: + return nil, fmt.Errorf("%w: unsupported control plane tls mode", ErrInvalidOptions) + } +} + +func validDialAddress(address string) bool { + host, port, err := net.SplitHostPort(address) + if err != nil || host == "" { + return false + } + value, err := strconv.ParseUint(port, 10, 16) + if err != nil || value == 0 { + return false + } + parsed := net.ParseIP(strings.Trim(host, "[]")) + return parsed == nil || !parsed.IsUnspecified() +} + +func loopbackAddress(address string) bool { + host, _, err := net.SplitHostPort(address) + if err != nil { + return false + } + host = strings.Trim(host, "[]") + if strings.EqualFold(host, "localhost") { + return true + } + ip := net.ParseIP(host) + return ip != nil && ip.IsLoopback() +} + +func nilInterface(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/gateway/bootstrap/bootstrap_test.go b/internal/gateway/bootstrap/bootstrap_test.go new file mode 100644 index 0000000..04ebee3 --- /dev/null +++ b/internal/gateway/bootstrap/bootstrap_test.go @@ -0,0 +1,296 @@ +package bootstrap + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "testing" + "time" + + controlplanev1 "proxy-pool/gen/controlplane/v1" + "proxy-pool/internal/config" + "proxy-pool/internal/controlplane/snapshotwire" + "proxy-pool/internal/gateway/snapshot" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/emptypb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestRunServesGatewayOnlyAfterApplyingControlPlaneSnapshot(t *testing.T) { + t.Parallel() + + target := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/through-gateway" { + t.Fatalf("target path = %q, want /through-gateway", request.URL.Path) + } + _, _ = io.WriteString(writer, "direct-route") + })) + t.Cleanup(target.Close) + + controlListener := mustListen(t) + controlServer := grpc.NewServer() + controlplanev1.RegisterWorkerControlPlaneServer(controlServer, &snapshotServer{ + snapshot: testSnapshot(t, target.URL), + }) + go func() { _ = controlServer.Serve(controlListener) }() + t.Cleanup(func() { + controlServer.Stop() + _ = controlListener.Close() + }) + + proxyListener := mustListen(t) + metricsListener := mustListen(t) + configPath := writeConfig(t, target.URL) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + result := make(chan error, 1) + go func() { + result <- Run(ctx, Options{ + ConfigPath: configPath, + Resolver: config.OSResolver{}, + ControlPlaneAddress: controlListener.Addr().String(), + ClusterID: "cluster-a", + WorkerID: "worker-a", + InstanceID: "instance-a", + Zone: "zone-a", + GatewayListener: proxyListener, + MetricsListener: metricsListener, + GRPCTransport: insecure.NewCredentials(), + ReconnectInitialDelay: 5 * time.Millisecond, + ReconnectMaxDelay: 20 * time.Millisecond, + }) + }() + + if err := waitForStatus("http://"+metricsListener.Addr().String()+"/readyz", http.StatusOK); err != nil { + select { + case runErr := <-result: + t.Fatalf("gateway did not become ready: %v; Run() error = %v", err, runErr) + default: + t.Fatalf("gateway did not become ready: %v", err) + } + } + proxyURL, err := url.Parse("http://" + proxyListener.Addr().String()) + if err != nil { + t.Fatalf("Parse(proxy URL): %v", err) + } + response, err := (&http.Client{Timeout: time.Second, Transport: &http.Transport{ + Proxy: http.ProxyURL(proxyURL), + }}).Get(target.URL + "/through-gateway") + if err != nil { + t.Fatalf("GET through gateway: %v", err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("ReadAll(gateway response): %v", err) + } + if response.StatusCode != http.StatusOK || string(body) != "direct-route" { + t.Fatalf("gateway response = (%d, %q)", response.StatusCode, body) + } + + cancel() + select { + case err := <-result: + if err != nil && !errors.Is(err, context.Canceled) { + t.Fatalf("Run() error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Run() did not stop after cancellation") + } +} + +func TestRunRejectsMissingControlPlaneEndpoint(t *testing.T) { + t.Parallel() + + err := Run(context.Background(), Options{ + ConfigPath: "config.yaml", Resolver: config.OSResolver{}, + ClusterID: "cluster-a", WorkerID: "worker-a", InstanceID: "instance-a", Zone: "zone-a", + }) + if !errors.Is(err, ErrInvalidOptions) { + t.Fatalf("Run() error = %v, want ErrInvalidOptions", err) + } +} + +func TestControlPlaneTransportRequiresDedicatedGatewayTLS(t *testing.T) { + t.Parallel() + + _, err := controlPlaneTransport(config.ControlPlane{ + TLS: config.ControlPlaneTLS{Mode: "mtls"}, + }, "controller.example:8443", nil) + if !errors.Is(err, ErrInvalidOptions) { + t.Fatalf("controlPlaneTransport() error = %v, want ErrInvalidOptions", err) + } +} + +func TestSnapshotReadinessRequiresCurrentSnapshot(t *testing.T) { + t.Parallel() + + store := snapshot.NewStore("cluster-a", "worker-a") + now := time.Now().UTC() + readiness := snapshotReadiness{store: store, now: func() time.Time { return now }} + if err := readiness.Ready(context.Background()); !errors.Is(err, ErrNotReady) { + t.Fatalf("Ready() before snapshot error = %v, want ErrNotReady", err) + } + + if err := store.Apply(snapshot.Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + Checksum: snapshot.ChecksumWithRouting(nil, nil), ValidUntil: now.Add(time.Second), + }); err != nil { + t.Fatalf("Apply() error = %v", err) + } + if err := readiness.Ready(context.Background()); err != nil { + t.Fatalf("Ready() after snapshot error = %v", err) + } + now = now.Add(2 * time.Second) + if err := readiness.Ready(context.Background()); !errors.Is(err, ErrNotReady) { + t.Fatalf("Ready() after expiry error = %v, want ErrNotReady", err) + } +} + +type snapshotServer struct { + controlplanev1.UnimplementedWorkerControlPlaneServer + snapshot *controlplanev1.WorkerSnapshot +} + +func (server *snapshotServer) RegisterWorker(_ context.Context, request *controlplanev1.RegisterWorkerRequest) (*controlplanev1.RegisterWorkerResponse, error) { + if request.GetWorkerId() != "worker-a" || request.GetInstanceId() != "instance-a" || request.GetZone() != "zone-a" { + return nil, errors.New("unexpected worker registration") + } + return &controlplanev1.RegisterWorkerResponse{ + WorkerId: request.GetWorkerId(), SessionId: "session-a", OwnershipEpoch: 1, + HeartbeatInterval: durationpb.New(time.Hour), MaxStaleAge: durationpb.New(2 * time.Hour), + }, nil +} + +func (server *snapshotServer) WatchSnapshots(_ *controlplanev1.WatchSnapshotsRequest, stream grpc.ServerStreamingServer[controlplanev1.SnapshotEnvelope]) error { + if err := stream.Send(&controlplanev1.SnapshotEnvelope{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: server.snapshot}}); err != nil { + return err + } + <-stream.Context().Done() + return stream.Context().Err() +} + +func (server *snapshotServer) AcknowledgeSnapshot(context.Context, *controlplanev1.AcknowledgeSnapshotRequest) (*emptypb.Empty, error) { + return &emptypb.Empty{}, nil +} + +func (server *snapshotServer) ReportRuntime(context.Context, *controlplanev1.ReportRuntimeRequest) (*controlplanev1.ReportRuntimeResponse, error) { + return &controlplanev1.ReportRuntimeResponse{AcceptedOwnershipEpoch: 1}, nil +} + +func testSnapshot(t *testing.T, _ string) *controlplanev1.WorkerSnapshot { + t.Helper() + generated := time.Now().UTC() + result := &controlplanev1.WorkerSnapshot{ + Version: 1, OwnershipEpoch: 1, GeneratedAt: timestamppb.New(generated), ValidUntil: timestamppb.New(generated.Add(time.Minute)), + Routing: []*controlplanev1.RoutingRule{{ + Name: "gateway-default", Enabled: true, HostRegex: ".+", Upstreams: []string{"provider-a"}, + Strategy: &controlplanev1.RoutingStrategy{Type: controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM}, + OnUnavailable: controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_DIRECT, + }}, + } + checksum, err := snapshotwire.Checksum(result) + if err != nil { + t.Fatalf("Checksum() error = %v", err) + } + result.Checksum = checksum[:] + return result +} + +func writeConfig(t *testing.T, target string) string { + t.Helper() + parsed, err := url.Parse(target) + if err != nil { + t.Fatalf("Parse(target URL): %v", err) + } + port := parsed.Port() + if port == "" { + t.Fatalf("target URL has no explicit port: %q", target) + } + path := filepath.Join(t.TempDir(), "gateway.yaml") + content := fmt.Sprintf(`version: 1 +security: + requireProtectionOnPublicListen: false +gateway: + enabled: true + listen: 127.0.0.1:0 + auth: {mode: none} + destinationPolicy: + denyPrivateNetworks: false + denyLoopback: false + denyLinkLocal: false + allowedPorts: [%s] +distribution: + enabled: false +admin: + enabled: false +controlPlane: + enabled: true + listen: 127.0.0.1:0 + protocolVersion: 1 + heartbeatInterval: 1s + sessionTTL: 3s + maxStaleAge: 2s + maxMessageBytes: 1048576 + maxRuntimeCounters: 10 + maxConcurrentStreams: 10 + tls: {mode: disabled} +metrics: + enabled: true + listen: 127.0.0.1:0 +upstreams: + provider-a: + enabled: true + exposure: [gateway] + provider: {billingMode: fetch, protocols: [http]} + api: {url: https://provider.invalid/api, method: GET, auth: {type: none}} + proxyAuth: {type: response} + pool: {maxSize: 1} + capacity: {maxConcurrencyPerProxy: 2} + refill: {reconcileInterval: 1s, minimumAvailableSlots: 1, targetAvailableSlots: 2} + lifecycle: {ttl: 1m, allocationSafetyMargin: 1s} + fetch: {estimatedIPsPerCall: 1, timeout: 1s, maxAttempts: 1, maxInFlight: 1} +`, port) + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatalf("WriteFile(): %v", err) + } + return path +} + +func mustListen(t *testing.T) net.Listener { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen(): %v", err) + } + return listener +} + +func waitForStatus(target string, want int) error { + deadline := time.Now().Add(2 * time.Second) + last := "no response" + for time.Now().Before(deadline) { + response, err := (&http.Client{Timeout: 100 * time.Millisecond}).Get(target) + if err == nil { + _ = response.Body.Close() + if response.StatusCode == want { + return nil + } + last = response.Status + } else { + last = err.Error() + } + time.Sleep(10 * time.Millisecond) + } + return fmt.Errorf("GET %s did not return %d: last result %s", target, want, last) +}