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)
+}