feat: distribute credentials in worker snapshots
Some checks are pending
ci / proto (push) Waiting to run
ci / test (ubuntu-latest) (push) Waiting to run
ci / test (windows-latest) (push) Waiting to run
ci / race (push) Waiting to run
ci / integration (push) Waiting to run

This commit is contained in:
youfak 2026-07-31 16:46:47 +08:00
parent 013defbd35
commit c267f77eee
19 changed files with 618 additions and 206 deletions

View File

@ -44,12 +44,12 @@ Proxy Pool 用 Controller 协调这些变化,并让 Gateway 数据面只消费
- **Gateway 组件**HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、重试、超时、
目的地址保护、本地快照存储、容量调度和 Worker 控制面 Register/Watch/ACK/
Runtime 会话组件已有实现与定向测试。`SessionSupervisor` 会为可恢复控制面中断执行
有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 及 Gateway
Routing 快照。Gateway 会将 Routing 与 Proxy 原子编译为同一内存 View并只按当前未过期
有界退避重连。Controller 可向 Worker 下发已归属 Proxy、Gateway Routing 与按引用去重的
凭据材料快照。Gateway 会将 Routing、Proxy 与凭据原子编译为同一内存 View并只按当前未过期
View 匹配请求,并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least
Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待,
以及仍经过目标地址策略的 direct`proxy-gateway` 通过独立控制面拨号地址维护
Session并仅在持有未过期 Snapshot 时 Ready凭据分发仍待装配
Session并仅在持有未过期 Snapshot 时 Ready凭据材料只保留在当前节点内存 View
- **安全边界**Gateway、Distribution 与 Admin 使用各自的认证语义,并支持
CIDR、可信代理、严格请求解析和敏感信息最小化。
@ -78,7 +78,7 @@ flowchart LR
## 当前完成度
截至 **2026-07-31**,实施计划检查项为 **54 / 7473.0%**。详情见
截至 **2026-07-31**,实施计划检查项为 **55 / 7474.3%**。详情见
[实施计划](docs/development/implementation-plan.md)和
[交付完成度审计](docs/requirements/completion-audit.md)。
@ -86,9 +86,9 @@ flowchart LR
提取与限流、Controller 的 Admin/Distribution/Metrics 监听,以及 PostgreSQL
管理状态WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和
Redis 会话栅栏。
- **部分完成**Gateway 的带凭据 Proxy 分发与 Outcome 上报、Checker 调度与健康状态链、
Docker Compose/Kubernetes 运行时 mTLS Overlay以及 protobuf 契约。
- **待完成**带凭据 Proxy 分发、Outcome 上报、Checker 调度与健康状态链,
- **部分完成**Gateway Outcome 上报、Checker 调度与健康状态链、Docker Compose/
Kubernetes 运行时 mTLS Overlay以及 protobuf 契约。
- **待完成**Outcome 上报、Checker 调度与健康状态链,
以及 loadgen 和代表性集群压测。
检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表
@ -185,8 +185,8 @@ Snapshot 时 `/readyz` 返回 `503`。Checker 与 loadgen 命令尚未实现。
ownership 索引,以及 Gateway 快照客户端。
- **P0 - Checker 健康链**Checker 调度、实际探测、Observation reducer以及
`FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链。
- **P1 - Gateway 与 Routing**Gateway 进程、五种 Routing 策略与
`onUnavailable` 已接入;带凭据 Proxy 分发、动态容量调整和 Drain 闭环待完成。
- **P1 - Gateway 与 Routing**Gateway 进程、快照凭据分发、五种 Routing 策略与
`onUnavailable` 已接入;动态容量调整和 Drain 闭环待完成。
- **P1 - 可观测与部署**:低基数业务指标、完整 Compose/Kubernetes 进程拓扑,
以及故障转移和恢复演练。
- **P2 - 容量证明**`proxy-loadgen`、HTTP/CONNECT/Extract 分场景压测,以及

View File

@ -64,6 +64,16 @@ message WorkerSnapshot {
bytes checksum = 5;
repeated RoutingRule routing = 6;
repeated OwnedProxy proxies = 7;
// Credential materials are protected by the WorkerControlPlane mTLS session
// and retained only in the Gateway's current in-memory snapshot view.
repeated SnapshotCredential credentials = 14;
}
message SnapshotCredential {
string secret_ref = 1;
string credential_version = 2;
string username = 3;
string password = 4;
}
message SnapshotDelta {

View File

@ -235,9 +235,9 @@ Provider Parser 继续通过 `credentials.Store` 生成 `SecretRef` 和
在活动池状态提交之前。Distribution 所需凭据只保存在 Proxy 硬 TTL 和幂等 TTL
约束内Extract 脚本可原子保存完整重放响应。
Gateway Snapshot 继续只携带凭据引用Gateway 通过控制面下发到节点内存的
凭据材料解析引用,不在请求热路径查询 Redis。凭据分发与轮换属于独立后续
实现,不改变本 ADR 的活动池边界。
Gateway Snapshot 携带凭据引用及按引用去重的材料;材料只经 mTLS 控制面进入
Gateway 当前内存 View不在请求热路径查询 Redis也不进入 PostgreSQL。凭据轮换的
主动推送仍属于后续实现,不改变本 ADR 的活动池边界。
### 库存真值

View File

@ -27,8 +27,9 @@ Controller 已实现并验证 `RegisterWorker`、`AcknowledgeSnapshot` 和
`WatchSnapshots` 已在 Register 后发送与当前 ownership epoch 对应的基础完整快照,
Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有界 Worker ownership
索引构建无凭据引用的已归属 Proxy 内容,并将租约到期收紧到 Proxy 的
`usable_until`;带凭据 Proxy 会 fail-closed直到凭据材料分发完成。完整 Snapshot
索引构建已归属 Proxy 内容,并将租约到期收紧到 Proxy 的 `usable_until`。Proxy 引用的
凭据材料按 `secret_ref + credential_version` 去重,随完整 Snapshot 经 mTLS 下发,仅保留在
Gateway 当前内存 View。完整 Snapshot
已从配置原始顺序和 Admin 当前状态合成 Gateway Routing并与 Proxy 一起纳入 checksum
Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本地 View动态 Router 只匹配
当前未过期 View派发器已按五种策略从该 View 选择上游,且在 Proxy 容量耗尽时只在该
@ -36,7 +37,7 @@ View 的其余候选中回退。`wait_timeout` 随 `on_unavailable=WAIT` 下发
本地容量等待使用;`DIRECT` 仍先经过 TargetPolicy 再建立 HTTP/CONNECT 直连。`proxy-gateway`
已装配 Register/Watch/ACK/Runtime 会话、HTTP 代理监听和 Snapshot 就绪探针;控制面中断时
保持进程运行并以有界退避重连,未取得有效 Snapshot 的 Worker 不会 Ready。增量、带凭据
Proxy 分发、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`
Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`
`100,000 QPS` 仍是未验证的设计目标。
`WatchSnapshots` 建立时校验当前 session每次签发快照引用时也把 `session_id`
@ -164,7 +165,8 @@ Routing 决定 AVAILABLE、SUSPECT 或 UNHEALTHY并更新 Redis 活动池,
集群环境使用 mTLS证书身份绑定 Worker/Checker 类型和环境。服务端校验
消息中的逻辑 ID 与证书授权一致,设置单消息大小、流持续时间、并发 Stream
和上报批次上限。`secret_ref` 是受控引用,不在 Proto 中传播真实密码。
和上报批次上限。`secret_ref` 是受控引用;完整 Snapshot 的 `credentials` 在 mTLS 会话中
携带引用对应材料Controller 和 Gateway 仅在内存处理,禁止写入 Redis/PostgreSQL、日志或指标。
## 9. Gateway 启动参数

View File

@ -214,8 +214,8 @@ WorkerControlPlane 现已接入 Controller 生命周期Register、ACK 和 Run
身份、消息/流限制和有界停机已实现。`WatchSnapshots` 会发送当前 epoch 的基础完整
Snapshot 并保持连接Gateway 已具备 Register/Watch/ACK/Runtime 会话协调组件与
`proxy-gateway` 进程装配。
按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本,并可构建无凭据引用的
已归属 Proxy payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成,
按 Worker 的可下发 ownership 索引已进入 Redis 原子脚本,并可构建已归属 Proxy、
去重凭据材料和 Routing 的完整 payload。Snapshot 签发与 session 匹配在同一 Redis Lua 操作中完成,
重注册会清除旧引用,避免迟到 Stream 覆盖新 session。Worker 服务端会在最近完整
Snapshot 的 `valid_until` 到达时结束流;公用 `SessionSupervisor` 已为 Gateway 调用方
提供可恢复错误的有界指数退避重连,并在参数/认证/协议错误时停止。Gateway Routing
@ -223,7 +223,7 @@ payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 check
Proxy 原子编译为同版本 View动态 Router 只匹配该未过期 View。派发器的五种上游选择已
接入该 View并在容量耗尽时在同版本候选中回退`onUnavailable` 的 reject、wait 与 direct
已接入 Gateway`proxy-gateway` 已装配本地 HTTP/Metrics 监听、快照就绪探针和
控制面重连,凭据分发与 Outcome 上报仍未实现。
控制面重连与快照凭据分发,Outcome 上报仍未实现。
已新增公用 `domain/activitypool` 契约及并发安全内存参考实现Provider
Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次;已覆盖
@ -260,7 +260,7 @@ Controller 多副本共享同一计数。Client 身份只以 SHA-256 摘要进
原子回收历史字段Redis 异常 fail-closed 并返回 503真实额度耗尽返回 429。
Gateway 请求热路径仍只使用本地准入,不增加 Redis/PostgreSQL 调用。
WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本、基础
Snapshot 流和 Gateway 会话客户端已完成;权威 Proxy/Routing 发布、凭据分发
Snapshot 流、Gateway 会话客户端与快照凭据分发已完成;权威 Proxy/Routing 发布
Outcome 和健康执行链仍待完成,因此 Task 10 尚未全部完成。
## Task 11: Checker and Health Reducer

View File

@ -53,9 +53,10 @@
服务端 TTL负向 ACK 会关闭 Runtime 写入栅栏。Controller 已装配 loopback
plaintext fixture 与 SPIFFE mTLS 服务端;基础 Snapshot 流和 Gateway 的
Register/Watch/ACK/Runtime 会话协调已实现。Redis 以 Worker 可下发 ownership
索引构建无凭据引用的已归属 Proxy payload并以租约收紧可用期Gateway Routing
payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum其运行时消费、
凭据分发、Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
索引构建已归属 Proxy payload并以租约收紧可用期凭据材料按引用去重并与 Proxy/
Routing 一起纳入 Snapshot checksum仅在 mTLS 和 Gateway 当前内存 View 中处理。Gateway
Routing payload 已由配置顺序和 Admin 当前状态合成并覆盖 Snapshot checksum其运行时消费、
Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前
`session_id`,重注册会清除旧引用,迟到旧 Stream 不会覆盖新 session。Controller
在最近成功下发的 Snapshot `valid_until` 到达时关闭流Gateway 的公用
`SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot
@ -101,9 +102,9 @@ CI 已配置 Linux race job。PostgreSQL 18 和 Redis 8.2 的隔离 Adapter fixt
5. Worker Register/ACK/Runtime gRPC 接收路径、Redis 服务端 TTL 会话生命周期、
SPIFFE mTLS 校验和 Controller 生命周期接线已完成Redis Provider Leader、
分布式请求额度、Distribution Client 限制和 Provider 状态丢失重建已完成。
6. Worker 基础网络快照流、无凭据引用 Proxy/Gateway Routing payload、Gateway Snapshot
6. Worker 基础网络快照流、Proxy/Gateway Routing/凭据 Snapshot payload、Gateway Snapshot
客户端和进程装配、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成;
凭据分发与 Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。
Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。
7. Checker 调度、探测器和健康 reducer。
8. Admin/Distribution 细粒度授权和审计查询Distribution 分布式限流已完成。
9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。

View File

@ -7,7 +7,7 @@
| ID | 最终需求 | 来源 | 验证证据 |
|---|---|---|---|
| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC并发布无凭据 Proxy/Gateway Routing 完整快照Gateway 已将同版本 Routing 编译为动态匹配 View并由独立进程维护控制面会话。Checker/Loadgen 构建产物与带凭据 Proxy 分发待实现 |
| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离Controller 已运行 Worker Register/Watch/ACK/Runtime gRPC并发布 Proxy/Gateway Routing/按引用去重凭据完整快照Gateway 已将它们编译为同版本动态 View并由独立进程维护控制面会话。Checker/Loadgen 构建产物待实现 |
| 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 的编排待完成 |

View File

@ -604,8 +604,11 @@ type WorkerSnapshot struct {
Checksum []byte `protobuf:"bytes,5,opt,name=checksum,proto3" json:"checksum,omitempty"`
Routing []*RoutingRule `protobuf:"bytes,6,rep,name=routing,proto3" json:"routing,omitempty"`
Proxies []*OwnedProxy `protobuf:"bytes,7,rep,name=proxies,proto3" json:"proxies,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// Credential materials are protected by the WorkerControlPlane mTLS session
// and retained only in the Gateway's current in-memory snapshot view.
Credentials []*SnapshotCredential `protobuf:"bytes,14,rep,name=credentials,proto3" json:"credentials,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *WorkerSnapshot) Reset() {
@ -687,6 +690,81 @@ func (x *WorkerSnapshot) GetProxies() []*OwnedProxy {
return nil
}
func (x *WorkerSnapshot) GetCredentials() []*SnapshotCredential {
if x != nil {
return x.Credentials
}
return nil
}
type SnapshotCredential struct {
state protoimpl.MessageState `protogen:"open.v1"`
SecretRef string `protobuf:"bytes,1,opt,name=secret_ref,json=secretRef,proto3" json:"secret_ref,omitempty"`
CredentialVersion string `protobuf:"bytes,2,opt,name=credential_version,json=credentialVersion,proto3" json:"credential_version,omitempty"`
Username string `protobuf:"bytes,3,opt,name=username,proto3" json:"username,omitempty"`
Password string `protobuf:"bytes,4,opt,name=password,proto3" json:"password,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *SnapshotCredential) Reset() {
*x = SnapshotCredential{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *SnapshotCredential) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*SnapshotCredential) ProtoMessage() {}
func (x *SnapshotCredential) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use SnapshotCredential.ProtoReflect.Descriptor instead.
func (*SnapshotCredential) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{5}
}
func (x *SnapshotCredential) GetSecretRef() string {
if x != nil {
return x.SecretRef
}
return ""
}
func (x *SnapshotCredential) GetCredentialVersion() string {
if x != nil {
return x.CredentialVersion
}
return ""
}
func (x *SnapshotCredential) GetUsername() string {
if x != nil {
return x.Username
}
return ""
}
func (x *SnapshotCredential) GetPassword() string {
if x != nil {
return x.Password
}
return ""
}
type SnapshotDelta struct {
state protoimpl.MessageState `protogen:"open.v1"`
BaseVersion uint64 `protobuf:"varint,1,opt,name=base_version,json=baseVersion,proto3" json:"base_version,omitempty"`
@ -704,7 +782,7 @@ type SnapshotDelta struct {
func (x *SnapshotDelta) Reset() {
*x = SnapshotDelta{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[5]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -716,7 +794,7 @@ func (x *SnapshotDelta) String() string {
func (*SnapshotDelta) ProtoMessage() {}
func (x *SnapshotDelta) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[5]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -729,7 +807,7 @@ func (x *SnapshotDelta) ProtoReflect() protoreflect.Message {
// Deprecated: Use SnapshotDelta.ProtoReflect.Descriptor instead.
func (*SnapshotDelta) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{5}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{6}
}
func (x *SnapshotDelta) GetBaseVersion() uint64 {
@ -813,7 +891,7 @@ type RoutingRule struct {
func (x *RoutingRule) Reset() {
*x = RoutingRule{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[6]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -825,7 +903,7 @@ func (x *RoutingRule) String() string {
func (*RoutingRule) ProtoMessage() {}
func (x *RoutingRule) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[6]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -838,7 +916,7 @@ func (x *RoutingRule) ProtoReflect() protoreflect.Message {
// Deprecated: Use RoutingRule.ProtoReflect.Descriptor instead.
func (*RoutingRule) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{6}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{7}
}
func (x *RoutingRule) GetName() string {
@ -922,7 +1000,7 @@ type RoutingStrategy struct {
func (x *RoutingStrategy) Reset() {
*x = RoutingStrategy{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[7]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -934,7 +1012,7 @@ func (x *RoutingStrategy) String() string {
func (*RoutingStrategy) ProtoMessage() {}
func (x *RoutingStrategy) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[7]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -947,7 +1025,7 @@ func (x *RoutingStrategy) ProtoReflect() protoreflect.Message {
// Deprecated: Use RoutingStrategy.ProtoReflect.Descriptor instead.
func (*RoutingStrategy) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{7}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{8}
}
func (x *RoutingStrategy) GetType() StrategyType {
@ -992,7 +1070,7 @@ type OwnedProxy struct {
func (x *OwnedProxy) Reset() {
*x = OwnedProxy{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[8]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1004,7 +1082,7 @@ func (x *OwnedProxy) String() string {
func (*OwnedProxy) ProtoMessage() {}
func (x *OwnedProxy) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[8]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1017,7 +1095,7 @@ func (x *OwnedProxy) ProtoReflect() protoreflect.Message {
// Deprecated: Use OwnedProxy.ProtoReflect.Descriptor instead.
func (*OwnedProxy) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{8}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{9}
}
func (x *OwnedProxy) GetId() string {
@ -1127,7 +1205,7 @@ type AcknowledgeSnapshotRequest struct {
func (x *AcknowledgeSnapshotRequest) Reset() {
*x = AcknowledgeSnapshotRequest{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[9]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[10]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1139,7 +1217,7 @@ func (x *AcknowledgeSnapshotRequest) String() string {
func (*AcknowledgeSnapshotRequest) ProtoMessage() {}
func (x *AcknowledgeSnapshotRequest) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[9]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[10]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1152,7 +1230,7 @@ func (x *AcknowledgeSnapshotRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use AcknowledgeSnapshotRequest.ProtoReflect.Descriptor instead.
func (*AcknowledgeSnapshotRequest) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{9}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{10}
}
func (x *AcknowledgeSnapshotRequest) GetWorkerId() string {
@ -1223,7 +1301,7 @@ type OutcomeBatch struct {
func (x *OutcomeBatch) Reset() {
*x = OutcomeBatch{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[10]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[11]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1235,7 +1313,7 @@ func (x *OutcomeBatch) String() string {
func (*OutcomeBatch) ProtoMessage() {}
func (x *OutcomeBatch) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[10]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[11]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1248,7 +1326,7 @@ func (x *OutcomeBatch) ProtoReflect() protoreflect.Message {
// Deprecated: Use OutcomeBatch.ProtoReflect.Descriptor instead.
func (*OutcomeBatch) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{10}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{11}
}
func (x *OutcomeBatch) GetWorkerId() string {
@ -1294,7 +1372,7 @@ type ProxyOutcome struct {
func (x *ProxyOutcome) Reset() {
*x = ProxyOutcome{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[11]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[12]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1306,7 +1384,7 @@ func (x *ProxyOutcome) String() string {
func (*ProxyOutcome) ProtoMessage() {}
func (x *ProxyOutcome) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[11]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[12]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1319,7 +1397,7 @@ func (x *ProxyOutcome) ProtoReflect() protoreflect.Message {
// Deprecated: Use ProxyOutcome.ProtoReflect.Descriptor instead.
func (*ProxyOutcome) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{11}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{12}
}
func (x *ProxyOutcome) GetProxyId() string {
@ -1380,7 +1458,7 @@ type ReportOutcomesResponse struct {
func (x *ReportOutcomesResponse) Reset() {
*x = ReportOutcomesResponse{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[12]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[13]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1392,7 +1470,7 @@ func (x *ReportOutcomesResponse) String() string {
func (*ReportOutcomesResponse) ProtoMessage() {}
func (x *ReportOutcomesResponse) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[12]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[13]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1405,7 +1483,7 @@ func (x *ReportOutcomesResponse) ProtoReflect() protoreflect.Message {
// Deprecated: Use ReportOutcomesResponse.ProtoReflect.Descriptor instead.
func (*ReportOutcomesResponse) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{12}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{13}
}
func (x *ReportOutcomesResponse) GetAcceptedThroughSequence() uint64 {
@ -1430,7 +1508,7 @@ type ReportRuntimeRequest struct {
func (x *ReportRuntimeRequest) Reset() {
*x = ReportRuntimeRequest{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[13]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[14]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1442,7 +1520,7 @@ func (x *ReportRuntimeRequest) String() string {
func (*ReportRuntimeRequest) ProtoMessage() {}
func (x *ReportRuntimeRequest) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[13]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[14]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1455,7 +1533,7 @@ func (x *ReportRuntimeRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use ReportRuntimeRequest.ProtoReflect.Descriptor instead.
func (*ReportRuntimeRequest) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{13}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{14}
}
func (x *ReportRuntimeRequest) GetWorkerId() string {
@ -1519,7 +1597,7 @@ type ProxyRuntime struct {
func (x *ProxyRuntime) Reset() {
*x = ProxyRuntime{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[14]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[15]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1531,7 +1609,7 @@ func (x *ProxyRuntime) String() string {
func (*ProxyRuntime) ProtoMessage() {}
func (x *ProxyRuntime) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[14]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[15]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1544,7 +1622,7 @@ func (x *ProxyRuntime) ProtoReflect() protoreflect.Message {
// Deprecated: Use ProxyRuntime.ProtoReflect.Descriptor instead.
func (*ProxyRuntime) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{14}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{15}
}
func (x *ProxyRuntime) GetProxyId() string {
@ -1586,7 +1664,7 @@ type ReportRuntimeResponse struct {
func (x *ReportRuntimeResponse) Reset() {
*x = ReportRuntimeResponse{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[15]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[16]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1598,7 +1676,7 @@ func (x *ReportRuntimeResponse) String() string {
func (*ReportRuntimeResponse) ProtoMessage() {}
func (x *ReportRuntimeResponse) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[15]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[16]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1611,7 +1689,7 @@ func (x *ReportRuntimeResponse) ProtoReflect() protoreflect.Message {
// Deprecated: Use ReportRuntimeResponse.ProtoReflect.Descriptor instead.
func (*ReportRuntimeResponse) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{15}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{16}
}
func (x *ReportRuntimeResponse) GetAcceptedOwnershipEpoch() uint64 {
@ -1647,7 +1725,7 @@ type StreamCheckTasksRequest struct {
func (x *StreamCheckTasksRequest) Reset() {
*x = StreamCheckTasksRequest{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[16]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[17]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1659,7 +1737,7 @@ func (x *StreamCheckTasksRequest) String() string {
func (*StreamCheckTasksRequest) ProtoMessage() {}
func (x *StreamCheckTasksRequest) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[16]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[17]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1672,7 +1750,7 @@ func (x *StreamCheckTasksRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use StreamCheckTasksRequest.ProtoReflect.Descriptor instead.
func (*StreamCheckTasksRequest) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{16}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{17}
}
func (x *StreamCheckTasksRequest) GetCheckerId() string {
@ -1723,7 +1801,7 @@ type CheckTask struct {
func (x *CheckTask) Reset() {
*x = CheckTask{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[17]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[18]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1735,7 +1813,7 @@ func (x *CheckTask) String() string {
func (*CheckTask) ProtoMessage() {}
func (x *CheckTask) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[17]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[18]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1748,7 +1826,7 @@ func (x *CheckTask) ProtoReflect() protoreflect.Message {
// Deprecated: Use CheckTask.ProtoReflect.Descriptor instead.
func (*CheckTask) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{17}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{18}
}
func (x *CheckTask) GetTaskId() string {
@ -1845,7 +1923,7 @@ type ObservationBatch struct {
func (x *ObservationBatch) Reset() {
*x = ObservationBatch{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[18]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[19]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1857,7 +1935,7 @@ func (x *ObservationBatch) String() string {
func (*ObservationBatch) ProtoMessage() {}
func (x *ObservationBatch) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[18]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[19]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1870,7 +1948,7 @@ func (x *ObservationBatch) ProtoReflect() protoreflect.Message {
// Deprecated: Use ObservationBatch.ProtoReflect.Descriptor instead.
func (*ObservationBatch) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{18}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{19}
}
func (x *ObservationBatch) GetCheckerId() string {
@ -1905,7 +1983,7 @@ type HealthObservation struct {
func (x *HealthObservation) Reset() {
*x = HealthObservation{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[19]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[20]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -1917,7 +1995,7 @@ func (x *HealthObservation) String() string {
func (*HealthObservation) ProtoMessage() {}
func (x *HealthObservation) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[19]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[20]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -1930,7 +2008,7 @@ func (x *HealthObservation) ProtoReflect() protoreflect.Message {
// Deprecated: Use HealthObservation.ProtoReflect.Descriptor instead.
func (*HealthObservation) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{19}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{20}
}
func (x *HealthObservation) GetTaskId() string {
@ -2013,7 +2091,7 @@ type ReportObservationsResponse struct {
func (x *ReportObservationsResponse) Reset() {
*x = ReportObservationsResponse{}
mi := &file_controlplane_v1_controlplane_proto_msgTypes[20]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[21]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@ -2025,7 +2103,7 @@ func (x *ReportObservationsResponse) String() string {
func (*ReportObservationsResponse) ProtoMessage() {}
func (x *ReportObservationsResponse) ProtoReflect() protoreflect.Message {
mi := &file_controlplane_v1_controlplane_proto_msgTypes[20]
mi := &file_controlplane_v1_controlplane_proto_msgTypes[21]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@ -2038,7 +2116,7 @@ func (x *ReportObservationsResponse) ProtoReflect() protoreflect.Message {
// Deprecated: Use ReportObservationsResponse.ProtoReflect.Descriptor instead.
func (*ReportObservationsResponse) Descriptor() ([]byte, []int) {
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{20}
return file_controlplane_v1_controlplane_proto_rawDescGZIP(), []int{21}
}
func (x *ReportObservationsResponse) GetAccepted() uint32 {
@ -2086,7 +2164,7 @@ const file_controlplane_v1_controlplane_proto_rawDesc = "" +
"\x10SnapshotEnvelope\x12@\n" +
"\x04full\x18\x01 \x01(\v2*.proxy_pool.controlplane.v1.WorkerSnapshotH\x00R\x04full\x12A\n" +
"\x05delta\x18\x02 \x01(\v2).proxy_pool.controlplane.v1.SnapshotDeltaH\x00R\x05deltaB\t\n" +
"\apayload\"\xf0\x02\n" +
"\apayload\"\xc2\x03\n" +
"\x0eWorkerSnapshot\x12\x18\n" +
"\aversion\x18\x01 \x01(\x04R\aversion\x12'\n" +
"\x0fownership_epoch\x18\x02 \x01(\x04R\x0eownershipEpoch\x12=\n" +
@ -2095,7 +2173,14 @@ const file_controlplane_v1_controlplane_proto_rawDesc = "" +
"validUntil\x12\x1a\n" +
"\bchecksum\x18\x05 \x01(\fR\bchecksum\x12A\n" +
"\arouting\x18\x06 \x03(\v2'.proxy_pool.controlplane.v1.RoutingRuleR\arouting\x12@\n" +
"\aproxies\x18\a \x03(\v2&.proxy_pool.controlplane.v1.OwnedProxyR\aproxies\"\xd7\x03\n" +
"\aproxies\x18\a \x03(\v2&.proxy_pool.controlplane.v1.OwnedProxyR\aproxies\x12P\n" +
"\vcredentials\x18\x0e \x03(\v2..proxy_pool.controlplane.v1.SnapshotCredentialR\vcredentials\"\x9a\x01\n" +
"\x12SnapshotCredential\x12\x1d\n" +
"\n" +
"secret_ref\x18\x01 \x01(\tR\tsecretRef\x12-\n" +
"\x12credential_version\x18\x02 \x01(\tR\x11credentialVersion\x12\x1a\n" +
"\busername\x18\x03 \x01(\tR\busername\x12\x1a\n" +
"\bpassword\x18\x04 \x01(\tR\bpassword\"\xd7\x03\n" +
"\rSnapshotDelta\x12!\n" +
"\fbase_version\x18\x01 \x01(\x04R\vbaseVersion\x12\x18\n" +
"\aversion\x18\x02 \x01(\x04R\aversion\x12'\n" +
@ -2295,7 +2380,7 @@ func file_controlplane_v1_controlplane_proto_rawDescGZIP() []byte {
}
var file_controlplane_v1_controlplane_proto_enumTypes = make([]protoimpl.EnumInfo, 5)
var file_controlplane_v1_controlplane_proto_msgTypes = make([]protoimpl.MessageInfo, 25)
var file_controlplane_v1_controlplane_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
var file_controlplane_v1_controlplane_proto_goTypes = []any{
(StrategyType)(0), // 0: proxy_pool.controlplane.v1.StrategyType
(UnavailableAction)(0), // 1: proxy_pool.controlplane.v1.UnavailableAction
@ -2307,87 +2392,89 @@ var file_controlplane_v1_controlplane_proto_goTypes = []any{
(*WatchSnapshotsRequest)(nil), // 7: proxy_pool.controlplane.v1.WatchSnapshotsRequest
(*SnapshotEnvelope)(nil), // 8: proxy_pool.controlplane.v1.SnapshotEnvelope
(*WorkerSnapshot)(nil), // 9: proxy_pool.controlplane.v1.WorkerSnapshot
(*SnapshotDelta)(nil), // 10: proxy_pool.controlplane.v1.SnapshotDelta
(*RoutingRule)(nil), // 11: proxy_pool.controlplane.v1.RoutingRule
(*RoutingStrategy)(nil), // 12: proxy_pool.controlplane.v1.RoutingStrategy
(*OwnedProxy)(nil), // 13: proxy_pool.controlplane.v1.OwnedProxy
(*AcknowledgeSnapshotRequest)(nil), // 14: proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest
(*OutcomeBatch)(nil), // 15: proxy_pool.controlplane.v1.OutcomeBatch
(*ProxyOutcome)(nil), // 16: proxy_pool.controlplane.v1.ProxyOutcome
(*ReportOutcomesResponse)(nil), // 17: proxy_pool.controlplane.v1.ReportOutcomesResponse
(*ReportRuntimeRequest)(nil), // 18: proxy_pool.controlplane.v1.ReportRuntimeRequest
(*ProxyRuntime)(nil), // 19: proxy_pool.controlplane.v1.ProxyRuntime
(*ReportRuntimeResponse)(nil), // 20: proxy_pool.controlplane.v1.ReportRuntimeResponse
(*StreamCheckTasksRequest)(nil), // 21: proxy_pool.controlplane.v1.StreamCheckTasksRequest
(*CheckTask)(nil), // 22: proxy_pool.controlplane.v1.CheckTask
(*ObservationBatch)(nil), // 23: proxy_pool.controlplane.v1.ObservationBatch
(*HealthObservation)(nil), // 24: proxy_pool.controlplane.v1.HealthObservation
(*ReportObservationsResponse)(nil), // 25: proxy_pool.controlplane.v1.ReportObservationsResponse
nil, // 26: proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry
nil, // 27: proxy_pool.controlplane.v1.RoutingRule.HeadersEntry
nil, // 28: proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry
nil, // 29: proxy_pool.controlplane.v1.OwnedProxy.TagsEntry
(*durationpb.Duration)(nil), // 30: google.protobuf.Duration
(*timestamppb.Timestamp)(nil), // 31: google.protobuf.Timestamp
(*emptypb.Empty)(nil), // 32: google.protobuf.Empty
(*SnapshotCredential)(nil), // 10: proxy_pool.controlplane.v1.SnapshotCredential
(*SnapshotDelta)(nil), // 11: proxy_pool.controlplane.v1.SnapshotDelta
(*RoutingRule)(nil), // 12: proxy_pool.controlplane.v1.RoutingRule
(*RoutingStrategy)(nil), // 13: proxy_pool.controlplane.v1.RoutingStrategy
(*OwnedProxy)(nil), // 14: proxy_pool.controlplane.v1.OwnedProxy
(*AcknowledgeSnapshotRequest)(nil), // 15: proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest
(*OutcomeBatch)(nil), // 16: proxy_pool.controlplane.v1.OutcomeBatch
(*ProxyOutcome)(nil), // 17: proxy_pool.controlplane.v1.ProxyOutcome
(*ReportOutcomesResponse)(nil), // 18: proxy_pool.controlplane.v1.ReportOutcomesResponse
(*ReportRuntimeRequest)(nil), // 19: proxy_pool.controlplane.v1.ReportRuntimeRequest
(*ProxyRuntime)(nil), // 20: proxy_pool.controlplane.v1.ProxyRuntime
(*ReportRuntimeResponse)(nil), // 21: proxy_pool.controlplane.v1.ReportRuntimeResponse
(*StreamCheckTasksRequest)(nil), // 22: proxy_pool.controlplane.v1.StreamCheckTasksRequest
(*CheckTask)(nil), // 23: proxy_pool.controlplane.v1.CheckTask
(*ObservationBatch)(nil), // 24: proxy_pool.controlplane.v1.ObservationBatch
(*HealthObservation)(nil), // 25: proxy_pool.controlplane.v1.HealthObservation
(*ReportObservationsResponse)(nil), // 26: proxy_pool.controlplane.v1.ReportObservationsResponse
nil, // 27: proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry
nil, // 28: proxy_pool.controlplane.v1.RoutingRule.HeadersEntry
nil, // 29: proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry
nil, // 30: proxy_pool.controlplane.v1.OwnedProxy.TagsEntry
(*durationpb.Duration)(nil), // 31: google.protobuf.Duration
(*timestamppb.Timestamp)(nil), // 32: google.protobuf.Timestamp
(*emptypb.Empty)(nil), // 33: google.protobuf.Empty
}
var file_controlplane_v1_controlplane_proto_depIdxs = []int32{
26, // 0: proxy_pool.controlplane.v1.RegisterWorkerRequest.labels:type_name -> proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry
30, // 1: proxy_pool.controlplane.v1.RegisterWorkerResponse.heartbeat_interval:type_name -> google.protobuf.Duration
30, // 2: proxy_pool.controlplane.v1.RegisterWorkerResponse.max_stale_age:type_name -> google.protobuf.Duration
27, // 0: proxy_pool.controlplane.v1.RegisterWorkerRequest.labels:type_name -> proxy_pool.controlplane.v1.RegisterWorkerRequest.LabelsEntry
31, // 1: proxy_pool.controlplane.v1.RegisterWorkerResponse.heartbeat_interval:type_name -> google.protobuf.Duration
31, // 2: proxy_pool.controlplane.v1.RegisterWorkerResponse.max_stale_age:type_name -> google.protobuf.Duration
9, // 3: proxy_pool.controlplane.v1.SnapshotEnvelope.full:type_name -> proxy_pool.controlplane.v1.WorkerSnapshot
10, // 4: proxy_pool.controlplane.v1.SnapshotEnvelope.delta:type_name -> proxy_pool.controlplane.v1.SnapshotDelta
31, // 5: proxy_pool.controlplane.v1.WorkerSnapshot.generated_at:type_name -> google.protobuf.Timestamp
31, // 6: proxy_pool.controlplane.v1.WorkerSnapshot.valid_until:type_name -> google.protobuf.Timestamp
11, // 7: proxy_pool.controlplane.v1.WorkerSnapshot.routing:type_name -> proxy_pool.controlplane.v1.RoutingRule
13, // 8: proxy_pool.controlplane.v1.WorkerSnapshot.proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy
31, // 9: proxy_pool.controlplane.v1.SnapshotDelta.generated_at:type_name -> google.protobuf.Timestamp
11, // 10: proxy_pool.controlplane.v1.SnapshotDelta.upserted_routing:type_name -> proxy_pool.controlplane.v1.RoutingRule
13, // 11: proxy_pool.controlplane.v1.SnapshotDelta.upserted_proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy
27, // 12: proxy_pool.controlplane.v1.RoutingRule.headers:type_name -> proxy_pool.controlplane.v1.RoutingRule.HeadersEntry
12, // 13: proxy_pool.controlplane.v1.RoutingRule.strategy:type_name -> proxy_pool.controlplane.v1.RoutingStrategy
1, // 14: proxy_pool.controlplane.v1.RoutingRule.on_unavailable:type_name -> proxy_pool.controlplane.v1.UnavailableAction
30, // 15: proxy_pool.controlplane.v1.RoutingRule.wait_timeout:type_name -> google.protobuf.Duration
0, // 16: proxy_pool.controlplane.v1.RoutingStrategy.type:type_name -> proxy_pool.controlplane.v1.StrategyType
28, // 17: proxy_pool.controlplane.v1.RoutingStrategy.weights:type_name -> proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry
2, // 18: proxy_pool.controlplane.v1.OwnedProxy.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol
31, // 19: proxy_pool.controlplane.v1.OwnedProxy.expires_at:type_name -> google.protobuf.Timestamp
29, // 20: proxy_pool.controlplane.v1.OwnedProxy.tags:type_name -> proxy_pool.controlplane.v1.OwnedProxy.TagsEntry
31, // 21: proxy_pool.controlplane.v1.OwnedProxy.usable_until:type_name -> google.protobuf.Timestamp
16, // 22: proxy_pool.controlplane.v1.OutcomeBatch.outcomes:type_name -> proxy_pool.controlplane.v1.ProxyOutcome
3, // 23: proxy_pool.controlplane.v1.ProxyOutcome.stage:type_name -> proxy_pool.controlplane.v1.OutcomeStage
30, // 24: proxy_pool.controlplane.v1.ProxyOutcome.latency:type_name -> google.protobuf.Duration
31, // 25: proxy_pool.controlplane.v1.ProxyOutcome.observed_at:type_name -> google.protobuf.Timestamp
19, // 26: proxy_pool.controlplane.v1.ReportRuntimeRequest.counters:type_name -> proxy_pool.controlplane.v1.ProxyRuntime
31, // 27: proxy_pool.controlplane.v1.ReportRuntimeRequest.observed_at:type_name -> google.protobuf.Timestamp
4, // 28: proxy_pool.controlplane.v1.StreamCheckTasksRequest.supported_levels:type_name -> proxy_pool.controlplane.v1.CheckLevel
2, // 29: proxy_pool.controlplane.v1.CheckTask.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol
4, // 30: proxy_pool.controlplane.v1.CheckTask.level:type_name -> proxy_pool.controlplane.v1.CheckLevel
30, // 31: proxy_pool.controlplane.v1.CheckTask.timeout:type_name -> google.protobuf.Duration
31, // 32: proxy_pool.controlplane.v1.CheckTask.deadline:type_name -> google.protobuf.Timestamp
24, // 33: proxy_pool.controlplane.v1.ObservationBatch.observations:type_name -> proxy_pool.controlplane.v1.HealthObservation
4, // 34: proxy_pool.controlplane.v1.HealthObservation.level:type_name -> proxy_pool.controlplane.v1.CheckLevel
30, // 35: proxy_pool.controlplane.v1.HealthObservation.latency:type_name -> google.protobuf.Duration
31, // 36: proxy_pool.controlplane.v1.HealthObservation.observed_at:type_name -> google.protobuf.Timestamp
5, // 37: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:input_type -> proxy_pool.controlplane.v1.RegisterWorkerRequest
7, // 38: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:input_type -> proxy_pool.controlplane.v1.WatchSnapshotsRequest
14, // 39: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:input_type -> proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest
15, // 40: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:input_type -> proxy_pool.controlplane.v1.OutcomeBatch
18, // 41: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:input_type -> proxy_pool.controlplane.v1.ReportRuntimeRequest
21, // 42: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:input_type -> proxy_pool.controlplane.v1.StreamCheckTasksRequest
23, // 43: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:input_type -> proxy_pool.controlplane.v1.ObservationBatch
6, // 44: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:output_type -> proxy_pool.controlplane.v1.RegisterWorkerResponse
8, // 45: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:output_type -> proxy_pool.controlplane.v1.SnapshotEnvelope
32, // 46: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:output_type -> google.protobuf.Empty
17, // 47: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:output_type -> proxy_pool.controlplane.v1.ReportOutcomesResponse
20, // 48: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:output_type -> proxy_pool.controlplane.v1.ReportRuntimeResponse
22, // 49: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:output_type -> proxy_pool.controlplane.v1.CheckTask
25, // 50: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:output_type -> proxy_pool.controlplane.v1.ReportObservationsResponse
44, // [44:51] is the sub-list for method output_type
37, // [37:44] is the sub-list for method input_type
37, // [37:37] is the sub-list for extension type_name
37, // [37:37] is the sub-list for extension extendee
0, // [0:37] is the sub-list for field type_name
11, // 4: proxy_pool.controlplane.v1.SnapshotEnvelope.delta:type_name -> proxy_pool.controlplane.v1.SnapshotDelta
32, // 5: proxy_pool.controlplane.v1.WorkerSnapshot.generated_at:type_name -> google.protobuf.Timestamp
32, // 6: proxy_pool.controlplane.v1.WorkerSnapshot.valid_until:type_name -> google.protobuf.Timestamp
12, // 7: proxy_pool.controlplane.v1.WorkerSnapshot.routing:type_name -> proxy_pool.controlplane.v1.RoutingRule
14, // 8: proxy_pool.controlplane.v1.WorkerSnapshot.proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy
10, // 9: proxy_pool.controlplane.v1.WorkerSnapshot.credentials:type_name -> proxy_pool.controlplane.v1.SnapshotCredential
32, // 10: proxy_pool.controlplane.v1.SnapshotDelta.generated_at:type_name -> google.protobuf.Timestamp
12, // 11: proxy_pool.controlplane.v1.SnapshotDelta.upserted_routing:type_name -> proxy_pool.controlplane.v1.RoutingRule
14, // 12: proxy_pool.controlplane.v1.SnapshotDelta.upserted_proxies:type_name -> proxy_pool.controlplane.v1.OwnedProxy
28, // 13: proxy_pool.controlplane.v1.RoutingRule.headers:type_name -> proxy_pool.controlplane.v1.RoutingRule.HeadersEntry
13, // 14: proxy_pool.controlplane.v1.RoutingRule.strategy:type_name -> proxy_pool.controlplane.v1.RoutingStrategy
1, // 15: proxy_pool.controlplane.v1.RoutingRule.on_unavailable:type_name -> proxy_pool.controlplane.v1.UnavailableAction
31, // 16: proxy_pool.controlplane.v1.RoutingRule.wait_timeout:type_name -> google.protobuf.Duration
0, // 17: proxy_pool.controlplane.v1.RoutingStrategy.type:type_name -> proxy_pool.controlplane.v1.StrategyType
29, // 18: proxy_pool.controlplane.v1.RoutingStrategy.weights:type_name -> proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry
2, // 19: proxy_pool.controlplane.v1.OwnedProxy.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol
32, // 20: proxy_pool.controlplane.v1.OwnedProxy.expires_at:type_name -> google.protobuf.Timestamp
30, // 21: proxy_pool.controlplane.v1.OwnedProxy.tags:type_name -> proxy_pool.controlplane.v1.OwnedProxy.TagsEntry
32, // 22: proxy_pool.controlplane.v1.OwnedProxy.usable_until:type_name -> google.protobuf.Timestamp
17, // 23: proxy_pool.controlplane.v1.OutcomeBatch.outcomes:type_name -> proxy_pool.controlplane.v1.ProxyOutcome
3, // 24: proxy_pool.controlplane.v1.ProxyOutcome.stage:type_name -> proxy_pool.controlplane.v1.OutcomeStage
31, // 25: proxy_pool.controlplane.v1.ProxyOutcome.latency:type_name -> google.protobuf.Duration
32, // 26: proxy_pool.controlplane.v1.ProxyOutcome.observed_at:type_name -> google.protobuf.Timestamp
20, // 27: proxy_pool.controlplane.v1.ReportRuntimeRequest.counters:type_name -> proxy_pool.controlplane.v1.ProxyRuntime
32, // 28: proxy_pool.controlplane.v1.ReportRuntimeRequest.observed_at:type_name -> google.protobuf.Timestamp
4, // 29: proxy_pool.controlplane.v1.StreamCheckTasksRequest.supported_levels:type_name -> proxy_pool.controlplane.v1.CheckLevel
2, // 30: proxy_pool.controlplane.v1.CheckTask.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol
4, // 31: proxy_pool.controlplane.v1.CheckTask.level:type_name -> proxy_pool.controlplane.v1.CheckLevel
31, // 32: proxy_pool.controlplane.v1.CheckTask.timeout:type_name -> google.protobuf.Duration
32, // 33: proxy_pool.controlplane.v1.CheckTask.deadline:type_name -> google.protobuf.Timestamp
25, // 34: proxy_pool.controlplane.v1.ObservationBatch.observations:type_name -> proxy_pool.controlplane.v1.HealthObservation
4, // 35: proxy_pool.controlplane.v1.HealthObservation.level:type_name -> proxy_pool.controlplane.v1.CheckLevel
31, // 36: proxy_pool.controlplane.v1.HealthObservation.latency:type_name -> google.protobuf.Duration
32, // 37: proxy_pool.controlplane.v1.HealthObservation.observed_at:type_name -> google.protobuf.Timestamp
5, // 38: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:input_type -> proxy_pool.controlplane.v1.RegisterWorkerRequest
7, // 39: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:input_type -> proxy_pool.controlplane.v1.WatchSnapshotsRequest
15, // 40: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:input_type -> proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest
16, // 41: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:input_type -> proxy_pool.controlplane.v1.OutcomeBatch
19, // 42: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:input_type -> proxy_pool.controlplane.v1.ReportRuntimeRequest
22, // 43: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:input_type -> proxy_pool.controlplane.v1.StreamCheckTasksRequest
24, // 44: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:input_type -> proxy_pool.controlplane.v1.ObservationBatch
6, // 45: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:output_type -> proxy_pool.controlplane.v1.RegisterWorkerResponse
8, // 46: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:output_type -> proxy_pool.controlplane.v1.SnapshotEnvelope
33, // 47: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:output_type -> google.protobuf.Empty
18, // 48: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:output_type -> proxy_pool.controlplane.v1.ReportOutcomesResponse
21, // 49: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:output_type -> proxy_pool.controlplane.v1.ReportRuntimeResponse
23, // 50: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:output_type -> proxy_pool.controlplane.v1.CheckTask
26, // 51: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:output_type -> proxy_pool.controlplane.v1.ReportObservationsResponse
45, // [45:52] is the sub-list for method output_type
38, // [38:45] is the sub-list for method input_type
38, // [38:38] is the sub-list for extension type_name
38, // [38:38] is the sub-list for extension extendee
0, // [0:38] is the sub-list for field type_name
}
func init() { file_controlplane_v1_controlplane_proto_init() }
@ -2405,7 +2492,7 @@ func file_controlplane_v1_controlplane_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_controlplane_v1_controlplane_proto_rawDesc), len(file_controlplane_v1_controlplane_proto_rawDesc)),
NumEnums: 5,
NumMessages: 25,
NumMessages: 26,
NumExtensions: 0,
NumServices: 2,
},

View File

@ -257,6 +257,7 @@ func runWithWorkerFactory(
MaxSnapshotBytes: loaded.Value.ControlPlane.MaxMessageBytes,
SnapshotReader: snapshotReader,
RoutingSource: routingSource,
Credentials: opened.credentials,
})
if serviceErr != nil {
return fmt.Errorf("%w: build Worker control service: %w", ErrStartup, serviceErr)

View File

@ -4,6 +4,7 @@ import (
"context"
"errors"
"math"
"sort"
"time"
controlplanev1 "proxy-pool/gen/controlplane/v1"
@ -11,6 +12,7 @@ import (
ownershipDomain "proxy-pool/internal/domain/ownership"
proxyDomain "proxy-pool/internal/domain/proxy"
"proxy-pool/internal/domain/workerruntime"
platformCredentials "proxy-pool/internal/platform/credentials"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
@ -21,13 +23,18 @@ var ErrSnapshotCredentialsUnavailable = errors.New("worker snapshot credential m
// OwnedSnapshotSource builds a complete Worker view from the Redis ownership
// index. It only accepts a complete bounded result from the reader.
type OwnedSnapshotSource struct {
epochs OwnershipEpochReader
reader ownershipDomain.SnapshotReader
validFor time.Duration
maxProxies int
maxBytes int
now func() time.Time
routing RoutingSource
epochs OwnershipEpochReader
reader ownershipDomain.SnapshotReader
validFor time.Duration
maxProxies int
maxBytes int
now func() time.Time
routing RoutingSource
credentials CredentialReader
}
type CredentialReader interface {
Resolve(context.Context, platformCredentials.Reference) (platformCredentials.Value, error)
}
func NewOwnedSnapshotSource(
@ -38,6 +45,35 @@ func NewOwnedSnapshotSource(
maxBytes int,
now func() time.Time,
routing ...RoutingSource,
) (*OwnedSnapshotSource, error) {
return newOwnedSnapshotSource(epochs, reader, validFor, maxProxies, maxBytes, now, nil, routing...)
}
func NewOwnedSnapshotSourceWithCredentials(
epochs OwnershipEpochReader,
reader ownershipDomain.SnapshotReader,
validFor time.Duration,
maxProxies int,
maxBytes int,
now func() time.Time,
credentials CredentialReader,
routing ...RoutingSource,
) (*OwnedSnapshotSource, error) {
if credentials == nil {
return nil, ErrSnapshotsUnavailable
}
return newOwnedSnapshotSource(epochs, reader, validFor, maxProxies, maxBytes, now, credentials, routing...)
}
func newOwnedSnapshotSource(
epochs OwnershipEpochReader,
reader ownershipDomain.SnapshotReader,
validFor time.Duration,
maxProxies int,
maxBytes int,
now func() time.Time,
credentials CredentialReader,
routing ...RoutingSource,
) (*OwnedSnapshotSource, error) {
if epochs == nil || reader == nil || validFor <= 0 || maxProxies <= 0 || maxBytes <= 0 || now == nil || len(routing) > 1 ||
len(routing) == 1 && routing[0] == nil {
@ -45,6 +81,7 @@ func NewOwnedSnapshotSource(
}
source := &OwnedSnapshotSource{
epochs: epochs, reader: reader, validFor: validFor, maxProxies: maxProxies, maxBytes: maxBytes, now: now,
credentials: credentials,
}
if len(routing) == 1 {
source.routing = routing[0]
@ -77,6 +114,7 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa
GeneratedAt: timestamppb.New(now), ValidUntil: timestamppb.New(now.Add(source.validFor)),
Proxies: make([]*controlplanev1.OwnedProxy, 0, len(proxies)),
}
materials := make(map[platformCredentials.Reference]platformCredentials.Value)
for _, item := range proxies {
owned, validUntil, err := wireOwnedProxy(item, now)
if err != nil {
@ -86,7 +124,26 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa
full.ValidUntil = timestamppb.New(validUntil)
}
full.Proxies = append(full.Proxies, owned)
if owned.GetSecretRef() != "" {
reference := platformCredentials.Reference{SecretRef: owned.GetSecretRef(), CredentialVersion: owned.GetCredentialVersion()}
if source.credentials == nil {
return nil, ErrSnapshotCredentialsUnavailable
}
material, resolveErr := source.credentials.Resolve(ctx, reference)
if resolveErr != nil || (material.Username == "" && material.Password == "") ||
(owned.GetUsername() != "" && material.Username != "" && material.Username != owned.GetUsername()) {
return nil, ErrSnapshotCredentialsUnavailable
}
if material.Username == "" {
material.Username = owned.GetUsername()
}
if existing, exists := materials[reference]; exists && existing != material {
return nil, ErrSnapshotCredentialsUnavailable
}
materials[reference] = material
}
}
full.Credentials = wireSnapshotCredentials(materials)
if source.routing != nil {
routing, err := source.routing.Read(ctx)
if err != nil {
@ -107,6 +164,23 @@ func (source *OwnedSnapshotSource) Watch(ctx context.Context, request SnapshotWa
return updates, nil
}
func wireSnapshotCredentials(materials map[platformCredentials.Reference]platformCredentials.Value) []*controlplanev1.SnapshotCredential {
result := make([]*controlplanev1.SnapshotCredential, 0, len(materials))
for reference, material := range materials {
result = append(result, &controlplanev1.SnapshotCredential{
SecretRef: reference.SecretRef, CredentialVersion: reference.CredentialVersion,
Username: material.Username, Password: material.Password,
})
}
sort.Slice(result, func(left, right int) bool {
if result[left].GetSecretRef() == result[right].GetSecretRef() {
return result[left].GetCredentialVersion() < result[right].GetCredentialVersion()
}
return result[left].GetSecretRef() < result[right].GetSecretRef()
})
return result
}
func cloneSnapshotRouting(source []*controlplanev1.RoutingRule) []*controlplanev1.RoutingRule {
result := make([]*controlplanev1.RoutingRule, len(source))
for index, rule := range source {
@ -119,7 +193,7 @@ func cloneSnapshotRouting(source []*controlplanev1.RoutingRule) []*controlplanev
func wireOwnedProxy(item ownershipDomain.SnapshotProxy, now time.Time) (*controlplanev1.OwnedProxy, time.Time, error) {
proxy := item.Proxy
if item.OwnershipEpoch == 0 || item.LeaseExpiresAt.IsZero() || proxy.CredentialVersion != "" || proxy.SecretRef != "" {
if item.OwnershipEpoch == 0 || item.LeaseExpiresAt.IsZero() || (proxy.CredentialVersion == "") != (proxy.SecretRef == "") {
if proxy.CredentialVersion != "" || proxy.SecretRef != "" {
return nil, time.Time{}, ErrSnapshotCredentialsUnavailable
}
@ -144,7 +218,7 @@ func wireOwnedProxy(item ownershipDomain.SnapshotProxy, now time.Time) (*control
}
return &controlplanev1.OwnedProxy{
Id: proxy.ID, Upstream: proxy.SourceUpstream, Protocol: protocol, Host: proxy.Host, Port: uint32(proxy.Port),
Username: proxy.Username, ExpiresAt: timestamppb.New(proxy.ExpiresAt.UTC()),
Username: proxy.Username, CredentialVersion: proxy.CredentialVersion, SecretRef: proxy.SecretRef, ExpiresAt: timestamppb.New(proxy.ExpiresAt.UTC()),
MaxConcurrency: uint32(proxy.MaxConcurrency), Tags: tags, OwnershipEpoch: item.OwnershipEpoch,
UsableUntil: timestamppb.New(validUntil),
}, validUntil, nil

View File

@ -10,6 +10,7 @@ import (
ownershipDomain "proxy-pool/internal/domain/ownership"
"proxy-pool/internal/domain/workerruntime"
platformCredentials "proxy-pool/internal/platform/credentials"
)
var (
@ -59,6 +60,7 @@ type Options struct {
MaxSnapshotBytes int
SnapshotReader ownershipDomain.SnapshotReader
RoutingSource RoutingSource
Credentials platformCredentials.Store
SessionID func() (string, error)
}
@ -136,9 +138,17 @@ func NewService(store workerruntime.ControlStore, options Options) (Service, err
if options.RoutingSource != nil {
routing = append(routing, options.RoutingSource)
}
source, err := NewOwnedSnapshotSource(
result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, routing...,
)
var source SnapshotSource
var err error
if options.Credentials != nil {
source, err = NewOwnedSnapshotSourceWithCredentials(
result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, options.Credentials, routing...,
)
} else {
source, err = NewOwnedSnapshotSource(
result, options.SnapshotReader, options.MaxStaleAge, options.MaxRuntimeCounters, options.MaxSnapshotBytes, time.Now, routing...,
)
}
if err != nil {
return nil, errors.Join(ErrInvalidCommand, err)
}

View File

@ -10,6 +10,7 @@ import (
"proxy-pool/internal/controlplane/snapshotwire"
ownershipDomain "proxy-pool/internal/domain/ownership"
proxyDomain "proxy-pool/internal/domain/proxy"
platformCredentials "proxy-pool/internal/platform/credentials"
)
func TestInitialSnapshotSourceIssuesNextFullSnapshot(t *testing.T) {
@ -123,6 +124,44 @@ func TestOwnedSnapshotSourceRejectsCredentialReferenceUntilMaterialIsAvailable(t
}
}
func TestOwnedSnapshotSourceIncludesReferencedCredentialMaterial(t *testing.T) {
now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC)
expiresAt := now.Add(time.Minute)
usableUntil := now.Add(30 * time.Second)
store, err := platformCredentials.NewMemoryStore(2)
if err != nil {
t.Fatalf("NewMemoryStore(): %v", err)
}
reference, err := store.Put(context.Background(), "provider-a", platformCredentials.Value{Username: "upstream", Password: "secret"})
if err != nil {
t.Fatalf("Put(): %v", err)
}
source, err := NewOwnedSnapshotSourceWithCredentials(epochReaderStub{epoch: 1}, snapshotReaderStub{proxies: []ownershipDomain.SnapshotProxy{{
Proxy: proxyDomain.Proxy{
ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, SourceUpstream: "upstream-a",
Username: "upstream", SecretRef: reference.SecretRef, CredentialVersion: reference.CredentialVersion,
ExpiresAt: &expiresAt, UsableUntil: &usableUntil, MaxConcurrency: 1, State: proxyDomain.StateAvailable,
}, OwnershipEpoch: 1, LeaseExpiresAt: now.Add(time.Minute),
}}}, time.Minute, 10, 4096, func() time.Time { return now }, store)
if err != nil {
t.Fatalf("NewOwnedSnapshotSourceWithCredentials(): %v", err)
}
updates, err := source.Watch(context.Background(), SnapshotWatchRequest{WorkerID: "worker-a", SessionID: "session-a"})
if err != nil {
t.Fatalf("Watch(): %v", err)
}
full := <-updates
if len(full.GetCredentials()) != 1 || full.GetCredentials()[0].GetSecretRef() != reference.SecretRef ||
full.GetCredentials()[0].GetUsername() != "upstream" || full.GetCredentials()[0].GetPassword() != "secret" ||
full.GetProxies()[0].GetCredentialVersion() != reference.CredentialVersion {
t.Fatal("snapshot credential material or proxy credential reference is invalid")
}
checksum, err := snapshotwire.Checksum(full)
if err != nil || string(checksum[:]) != string(full.GetChecksum()) {
t.Fatalf("snapshot checksum = %x, %v; want %x", full.GetChecksum(), err, checksum)
}
}
type epochReaderStub struct {
epoch uint64
err error

View File

@ -23,6 +23,7 @@ import (
controlplanev1 "proxy-pool/gen/controlplane/v1"
"proxy-pool/internal/config"
proxyDomain "proxy-pool/internal/domain/proxy"
"proxy-pool/internal/domain/workerruntime"
"proxy-pool/internal/gateway/controlplane"
"proxy-pool/internal/gateway/dispatch"
@ -145,7 +146,7 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio
if err != nil {
return nil, fmt.Errorf("build gateway target policy: %w", err)
}
proxyTransport := transport.New(transport.Config{}, nil)
proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store})
handler, err := server.New(server.ConfigFromListener(configuration.Gateway), server.Dependencies{
Auth: protection.Auth,
Access: protection.Access,
@ -222,6 +223,18 @@ type generatedRuntimeClient struct {
client controlplanev1.WorkerControlPlaneClient
}
type snapshotCredentialResolver struct {
store *snapshot.Store
}
func (resolver snapshotCredentialResolver) Resolve(ctx context.Context, selected proxyDomain.Proxy) (transport.Credentials, error) {
credential, err := resolver.store.Credential(ctx, selected)
if err != nil {
return transport.Credentials{}, err
}
return transport.Credentials{Username: credential.Username, Password: credential.Password}, nil
}
func (client generatedRuntimeClient) RegisterWorker(
ctx context.Context,
request *controlplanev1.RegisterWorkerRequest,

View File

@ -126,6 +126,10 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e
if err != nil {
return err
}
credentials, err := wireSnapshotCredentials(full.GetCredentials(), proxies)
if err != nil {
return err
}
routes, err := wireRouting(full.GetRouting())
if err != nil {
return err
@ -133,12 +137,42 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e
envelope := snapshot.Envelope{
ClusterID: watcher.options.ClusterID, WorkerID: watcher.options.WorkerID,
Epoch: full.GetOwnershipEpoch(), Version: full.GetVersion(), Full: true, ValidUntil: validUntil,
Proxies: proxies, Routing: routes,
Proxies: proxies, Routing: routes, Credentials: credentials,
}
envelope.Checksum = snapshot.ChecksumWithRouting(proxies, routes)
envelope.Checksum = snapshot.ChecksumWithCredentials(proxies, routes, credentials)
return watcher.store.Apply(envelope)
}
func wireSnapshotCredentials(source []*controlplanev1.SnapshotCredential, proxies []proxyDomain.Proxy) ([]snapshot.Credential, error) {
result := make([]snapshot.Credential, len(source))
indexed := make(map[string]struct{}, len(source))
for index, item := range source {
if item == nil || !workerruntime.ValidIdentifier(item.GetSecretRef()) || !workerruntime.ValidIdentifier(item.GetCredentialVersion()) ||
(item.GetUsername() == "" && item.GetPassword() == "") {
return nil, ErrInvalidSnapshotWatcher
}
key := item.GetSecretRef() + "\x00" + item.GetCredentialVersion()
if _, duplicate := indexed[key]; duplicate {
return nil, ErrInvalidSnapshotWatcher
}
indexed[key] = struct{}{}
result[index] = snapshot.Credential{
SecretRef: item.GetSecretRef(), CredentialVersion: item.GetCredentialVersion(),
Username: item.GetUsername(), Password: item.GetPassword(),
}
}
for _, proxy := range proxies {
if proxy.SecretRef == "" && proxy.CredentialVersion == "" {
continue
}
key := proxy.SecretRef + "\x00" + proxy.CredentialVersion
if _, exists := indexed[key]; !exists {
return nil, ErrInvalidSnapshotWatcher
}
}
return result, nil
}
func wireRouting(source []*controlplanev1.RoutingRule) ([]routingDomain.Rule, error) {
result := make([]routingDomain.Rule, 0, len(source))
names := make(map[string]struct{}, len(source))

View File

@ -23,7 +23,11 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T)
Version: 1, OwnershipEpoch: 7, GeneratedAt: timestamppb.New(time.Now()), ValidUntil: timestamppb.New(time.Now().Add(time.Minute)),
Proxies: []*controlplanev1.OwnedProxy{{
Id: "proxy-a", Upstream: "upstream-a", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP,
Host: "192.0.2.10", Port: 8080, MaxConcurrency: 3, ExpiresAt: timestamppb.New(time.Now().Add(time.Minute)),
Host: "192.0.2.10", Port: 8080, Username: "upstream", SecretRef: "cred_a", CredentialVersion: "v1",
MaxConcurrency: 3, ExpiresAt: timestamppb.New(time.Now().Add(time.Minute)),
}},
Credentials: []*controlplanev1.SnapshotCredential{{
SecretRef: "cred_a", CredentialVersion: "v1", Username: "upstream", Password: "secret",
}},
Routing: []*controlplanev1.RoutingRule{{
Name: "gateway-api", Enabled: true, HostRegex: `^api\.example\.test$`, Upstreams: []string{"upstream-a"},
@ -54,6 +58,10 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T)
if client.watch.GetSessionId() != "session-a" || client.ack.GetVersion() != 1 || !client.ack.GetApplied() || string(client.ack.GetChecksum()) != string(full.GetChecksum()) {
t.Fatalf("watch=%+v ack=%+v", client.watch, client.ack)
}
credential, err := store.Credential(context.Background(), view.Entries[0].Proxy)
if err != nil || credential.Username != "upstream" || credential.Password != "secret" {
t.Fatalf("Credential() = (%+v, %v)", credential, err)
}
}
func TestSnapshotWatcherRejectsChecksumAndAcknowledgesFailure(t *testing.T) {

View File

@ -277,7 +277,6 @@ func proxyDescriptor(t *testing.T, id, rawURL string) proxyDomain.Proxy {
SourceUpstream: "provider-a",
MaxConcurrency: 2,
State: proxyDomain.StateAvailable,
CredentialVersion: "v1",
}
}

View File

@ -595,7 +595,6 @@ func dispatcherWithProxies(t *testing.T, ids ...string) (*dispatch.Dispatcher, *
SourceUpstream: "provider-a",
MaxConcurrency: 2,
State: proxyDomain.StateAvailable,
CredentialVersion: "v1",
})
}
envelope := snapshot.Envelope{

View File

@ -1,6 +1,7 @@
package snapshot
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
@ -24,6 +25,8 @@ var (
ErrInvalidRuntimeReport = errors.New("invalid worker runtime report")
ErrInvalidRuntimeLimit = errors.New("invalid snapshot runtime limit")
ErrRuntimeLimitExceeded = errors.New("snapshot runtime limit exceeded")
ErrInvalidCredential = errors.New("invalid snapshot credential")
ErrCredentialMissing = errors.New("snapshot credential is missing")
)
const defaultRuntimeLimit = 1_000_000
@ -93,15 +96,29 @@ type runtimeRegistration struct {
}
type Envelope struct {
ClusterID string
WorkerID string
Epoch uint64
Version uint64
Full bool
Checksum string
ValidUntil time.Time
Proxies []proxyDomain.Proxy
Routing []routing.Rule
ClusterID string
WorkerID string
Epoch uint64
Version uint64
Full bool
Checksum string
ValidUntil time.Time
Proxies []proxyDomain.Proxy
Routing []routing.Rule
Credentials []Credential
}
// Credential is ephemeral material received through a verified mTLS snapshot.
// Its Format method prevents accidental diagnostics from exposing either value.
type Credential struct {
SecretRef string
CredentialVersion string
Username string
Password string
}
func (Credential) Format(state fmt.State, _ rune) {
_, _ = state.Write([]byte("snapshot.Credential{SecretRef:<redacted>, CredentialVersion:<redacted>, Username:<redacted>, Password:<redacted>}"))
}
type Entry struct {
@ -110,14 +127,15 @@ type Entry struct {
}
type View struct {
ClusterID string
WorkerID string
Epoch uint64
Version uint64
Checksum string
ValidUntil time.Time
Entries []Entry
routing *routing.RuleSet
ClusterID string
WorkerID string
Epoch uint64
Version uint64
Checksum string
ValidUntil time.Time
Entries []Entry
routing *routing.RuleSet
credentials map[string]Credential
all []int
byScheme map[proxyDomain.Scheme][]int
@ -229,13 +247,17 @@ func (s *Store) Apply(envelope Envelope) error {
if !envelope.Full || envelope.Epoch == 0 || envelope.Version == 0 {
return ErrResyncRequired
}
if envelope.Checksum != ChecksumWithRouting(envelope.Proxies, envelope.Routing) {
if envelope.Checksum != ChecksumWithCredentials(envelope.Proxies, envelope.Routing, envelope.Credentials) {
return ErrChecksumMismatch
}
routes, err := routing.Compile(envelope.Routing)
if err != nil {
return fmt.Errorf("compile snapshot routing: %w", err)
}
credentials, err := indexCredentials(envelope.Credentials, envelope.Proxies)
if err != nil {
return err
}
s.mu.Lock()
defer s.mu.Unlock()
@ -315,20 +337,56 @@ func (s *Store) Apply(envelope Envelope) error {
}
next := &View{
ClusterID: envelope.ClusterID,
WorkerID: envelope.WorkerID,
Epoch: envelope.Epoch,
Version: envelope.Version,
Checksum: envelope.Checksum,
ValidUntil: envelope.ValidUntil.UTC(),
Entries: entries,
routing: routes,
ClusterID: envelope.ClusterID,
WorkerID: envelope.WorkerID,
Epoch: envelope.Epoch,
Version: envelope.Version,
Checksum: envelope.Checksum,
ValidUntil: envelope.ValidUntil.UTC(),
Entries: entries,
routing: routes,
credentials: credentials,
}
next.buildIndexes()
s.current.Store(next)
return nil
}
// Credential resolves one exact Proxy credential reference from the current
// immutable view. Credentialless proxies preserve their metadata username.
func (s *Store) Credential(ctx context.Context, selected proxyDomain.Proxy) (Credential, error) {
if ctx == nil {
return Credential{}, context.Canceled
}
if err := ctx.Err(); err != nil {
return Credential{}, err
}
if (selected.SecretRef == "") != (selected.CredentialVersion == "") {
return Credential{}, ErrInvalidCredential
}
if selected.SecretRef == "" {
return Credential{Username: selected.Username}, nil
}
if s == nil {
return Credential{}, ErrCredentialMissing
}
current := s.current.Load()
if current == nil {
return Credential{}, ErrCredentialMissing
}
credential, ok := current.credentials[credentialKey(selected.SecretRef, selected.CredentialVersion)]
if !ok {
return Credential{}, ErrCredentialMissing
}
if selected.Username != "" && credential.Username != "" && selected.Username != credential.Username {
return Credential{}, ErrInvalidCredential
}
if credential.Username == "" {
credential.Username = selected.Username
}
return credential, nil
}
// MatchRouting matches a request against the immutable rules published with
// this proxy view. A View is the atomic consistency boundary for both sets.
func (v *View) MatchRouting(request routing.Request) (routing.Rule, bool) {
@ -427,12 +485,18 @@ func Checksum(proxies []proxyDomain.Proxy) string {
}
func ChecksumWithRouting(proxies []proxyDomain.Proxy, rules []routing.Rule) string {
return ChecksumWithCredentials(proxies, rules, nil)
}
func ChecksumWithCredentials(proxies []proxyDomain.Proxy, rules []routing.Rule, credentials []Credential) string {
payload := struct {
Proxies []proxyDomain.Proxy `json:"proxies"`
Routing []routing.Rule `json:"routing"`
Proxies []proxyDomain.Proxy `json:"proxies"`
Routing []routing.Rule `json:"routing"`
Credentials []Credential `json:"credentials"`
}{
Proxies: cloneAndSort(proxies),
Routing: cloneRoutingRules(rules),
Proxies: cloneAndSort(proxies),
Routing: cloneRoutingRules(rules),
Credentials: cloneAndSortCredentials(credentials),
}
encoded, err := json.Marshal(payload)
if err != nil {
@ -442,6 +506,46 @@ func ChecksumWithRouting(proxies []proxyDomain.Proxy, rules []routing.Rule) stri
return hex.EncodeToString(digest[:])
}
func indexCredentials(source []Credential, proxies []proxyDomain.Proxy) (map[string]Credential, error) {
indexed := make(map[string]Credential, len(source))
for _, credential := range source {
if credential.SecretRef == "" || credential.CredentialVersion == "" || (credential.Username == "" && credential.Password == "") {
return nil, ErrInvalidCredential
}
key := credentialKey(credential.SecretRef, credential.CredentialVersion)
if _, duplicate := indexed[key]; duplicate {
return nil, ErrInvalidCredential
}
indexed[key] = credential
}
for _, proxy := range proxies {
if (proxy.SecretRef == "") != (proxy.CredentialVersion == "") {
return nil, ErrInvalidCredential
}
if proxy.SecretRef == "" {
continue
}
credential, exists := indexed[credentialKey(proxy.SecretRef, proxy.CredentialVersion)]
if !exists || (proxy.Username != "" && credential.Username != "" && proxy.Username != credential.Username) {
return nil, ErrInvalidCredential
}
}
return indexed, nil
}
func credentialKey(secretRef, version string) string { return secretRef + "\x00" + version }
func cloneAndSortCredentials(source []Credential) []Credential {
cloned := append([]Credential(nil), source...)
sort.Slice(cloned, func(left, right int) bool {
if cloned[left].SecretRef == cloned[right].SecretRef {
return cloned[left].CredentialVersion < cloned[right].CredentialVersion
}
return cloned[left].SecretRef < cloned[right].SecretRef
})
return cloned
}
func cloneRoutingRules(source []routing.Rule) []routing.Rule {
result := make([]routing.Rule, len(source))
for index, rule := range source {

View File

@ -1,9 +1,11 @@
package snapshot
import (
"context"
"errors"
"fmt"
"reflect"
"strings"
"testing"
"time"
@ -126,6 +128,35 @@ func TestStorePublishesRoutingWithTheSameSnapshotVersion(t *testing.T) {
}
}
func TestStoreResolvesCredentialOnlyFromCurrentSnapshot(t *testing.T) {
store := NewStore("cluster-a", "worker-a")
proxy := proxyDomain.Proxy{
ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080, MaxConcurrency: 1,
SecretRef: "cred_a", CredentialVersion: "v1", Username: "upstream",
}
envelope := Envelope{
ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, Proxies: []proxyDomain.Proxy{proxy},
Credentials: []Credential{{SecretRef: "cred_a", CredentialVersion: "v1", Username: "upstream", Password: "secret"}},
}
envelope.Checksum = ChecksumWithCredentials(envelope.Proxies, envelope.Routing, envelope.Credentials)
if err := store.Apply(envelope); err != nil {
t.Fatalf("Apply(): %v", err)
}
credential, err := store.Credential(context.Background(), proxy)
if err != nil || credential.Username != "upstream" || credential.Password != "secret" {
t.Fatalf("Credential() = (%+v, %v)", credential, err)
}
if formatted := fmt.Sprintf("%+v", credential); strings.Contains(formatted, "secret") || strings.Contains(formatted, "upstream") {
t.Fatalf("formatted credential exposes material: %s", formatted)
}
missing := Envelope{ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 2, Full: true, Proxies: []proxyDomain.Proxy{proxy}}
missing.Checksum = ChecksumWithCredentials(missing.Proxies, nil, nil)
if err := store.Apply(missing); !errors.Is(err, ErrInvalidCredential) {
t.Fatalf("Apply(missing credential) error = %v, want ErrInvalidCredential", err)
}
}
func TestViewSelectFiltersBySchemeUpstreamTagAndExclude(t *testing.T) {
now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC)
expiresSoon := now.Add(5 * time.Second)