feat: handle unavailable gateway routes
This commit is contained in:
youfak
parent
a5b456a9d0
commit
265026a907
@ -46,7 +46,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
|
||||
有界退避重连。Controller 可向 Worker 下发无凭据引用的已归属 Proxy 及 Gateway
|
||||
Routing 快照。Gateway 会将 Routing 与 Proxy 原子编译为同一内存 View，并只按当前未过期
|
||||
View 匹配请求，并在内存中按 Sequential、Random、Round Robin、Weighted 或 Least
|
||||
Connections 选择上游；`proxy-gateway` 命令、凭据分发和 `onUnavailable` 派发语义仍待装配。
|
||||
Connections 选择上游。无候选时支持 reject、受 `waitTimeout` 限制的本地容量等待，
|
||||
以及仍经过目标地址策略的 direct；`proxy-gateway` 命令与凭据分发仍待装配。
|
||||
- **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持
|
||||
CIDR、可信代理、严格请求解析和敏感信息最小化。
|
||||
|
||||
@ -85,8 +86,8 @@ flowchart LR
|
||||
Redis 会话栅栏。
|
||||
- **部分完成**：Gateway 传输与调度组件、Snapshot 本地存储、Worker ownership
|
||||
与运行态领域组件、Docker Compose/Kubernetes 静态部署清单和 protobuf 契约。
|
||||
- **待完成**：带凭据 Proxy 分发、Gateway 进程装配、`onUnavailable`、Outcome 上报、
|
||||
Checker 调度与健康状态链，以及 loadgen 和代表性集群压测。
|
||||
- **待完成**：带凭据 Proxy 分发、Gateway 进程装配、Outcome 上报、Checker 调度与健康状态链，
|
||||
以及 loadgen 和代表性集群压测。
|
||||
|
||||
检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表
|
||||
端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。
|
||||
|
||||
@ -88,6 +88,7 @@ message RoutingRule {
|
||||
repeated string upstreams = 7;
|
||||
RoutingStrategy strategy = 8;
|
||||
UnavailableAction on_unavailable = 9;
|
||||
google.protobuf.Duration wait_timeout = 10;
|
||||
}
|
||||
|
||||
message RoutingStrategy {
|
||||
|
||||
@ -32,8 +32,10 @@ Gateway 校验后 ACK 并开始 Runtime 心跳。Controller 会从 Redis 的有
|
||||
已从配置原始顺序和 Admin 当前状态合成 Gateway Routing，并与 Proxy 一起纳入 checksum；
|
||||
Gateway 已将该 payload 编译并原子发布到与 Proxy 相同版本的本地 View，动态 Router 只匹配
|
||||
当前未过期 View；派发器已按五种策略从该 View 选择上游，且在 Proxy 容量耗尽时只在该
|
||||
View 的其余候选中回退。`onUnavailable`、增量、Gateway 进程装配、Outcome 与 Checker 闭环
|
||||
尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；`100,000 QPS` 仍是未验证的设计目标。
|
||||
View 的其余候选中回退。`wait_timeout` 随 `on_unavailable=WAIT` 下发并在 Gateway 作为有界
|
||||
本地容量等待使用；`DIRECT` 仍先经过 TargetPolicy 再建立 HTTP/CONNECT 直连。增量、Gateway
|
||||
进程装配、Outcome 与 Checker 闭环尚未实现。`ReportOutcomes` 仍明确返回 `Unimplemented`；
|
||||
`100,000 QPS` 仍是未验证的设计目标。
|
||||
|
||||
`WatchSnapshots` 建立时校验当前 session；每次签发快照引用时也把 `session_id`
|
||||
交给 Redis 原子校验。重复 Register 会同时清除旧 Runtime 和已签发引用，因此迟到的
|
||||
|
||||
@ -219,9 +219,9 @@ Snapshot 并保持连接；Gateway 已具备 Register/Watch/ACK/Runtime 会话
|
||||
Snapshot 的 `valid_until` 到达时结束流；公用 `SessionSupervisor` 已为 Gateway 调用方
|
||||
提供可恢复错误的有界指数退避重连，并在参数/认证/协议错误时停止。Gateway Routing
|
||||
payload 已按配置顺序和 Admin revision/current 状态发布并覆盖 checksum；Gateway 已将其与
|
||||
Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。策略选择、`onUnavailable`、
|
||||
派发器的五种上游选择已接入该 View，并在容量耗尽时在同版本候选中回退；`onUnavailable`、
|
||||
凭据分发、Gateway 命令与 Outcome 上报仍未实现。
|
||||
Proxy 原子编译为同版本 View，动态 Router 只匹配该未过期 View。派发器的五种上游选择已
|
||||
接入该 View，并在容量耗尽时在同版本候选中回退；`onUnavailable` 的 reject、wait 与 direct
|
||||
已接入 Gateway，凭据分发、Gateway 命令与 Outcome 上报仍未实现。
|
||||
|
||||
已新增公用 `domain/activitypool` 契约及并发安全内存参考实现，Provider
|
||||
Reconciler 通过 `UpsertFetched` 写入带供应商 TTL 和分配安全余量的批次；已覆盖
|
||||
|
||||
@ -100,13 +100,13 @@ 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 编译/动态匹配和五种策略上游选择已完成；`onUnavailable`、凭据
|
||||
客户端、同版本 Routing 编译/动态匹配、五种策略上游选择与 reject/wait/direct 已完成；凭据
|
||||
分发与 Outcome 上报仍待完成。Redis ownership drain/ACK/过期回收及按 Worker 的可下发索引已完成。
|
||||
7. Checker 调度、探测器和健康 reducer。
|
||||
8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。
|
||||
9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。
|
||||
10. 将 `onUnavailable` 接入 Gateway/Distribution 运行链，补齐 Sequential 持久化恢复、
|
||||
跨实例 CAS 和 disabled candidate 语义。
|
||||
10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS
|
||||
和 disabled candidate 语义。
|
||||
11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测；短 TTL
|
||||
Proxy 的零计数运行态已在后续 Snapshot Apply 中回收。
|
||||
|
||||
|
||||
@ -23,7 +23,7 @@
|
||||
| ROUTE-004 | Sequential 连续空结果达到阈值后原子切换一次 | 5295-5824, 6520-6617 | 进程内 `RoutingCursor` 版本 CAS 与 100 并发测试已完成；持久化恢复和跨实例 CAS 待完成 |
|
||||
| ROUTE-005 | 空计数属于 Upstream，当前选择属于 Routing | 8442-8529 | 共享 `UpstreamEmptyState` 双 Routing 测试 |
|
||||
| ROUTE-006 | 旧 Upstream 已有 Proxy 继续耗尽，不因切换直接丢弃 | 6618-6641 | 通用 ownership Drain/ACK 原语已测试；Routing 切换到 Drain 的编排待完成 |
|
||||
| ROUTE-007 | 无可用 Upstream 时显式 reject、wait 或 direct，默认 reject | 5075-5294, 6743-6760 | 三种动作的配置校验已完成；Gateway 当前仅返回 503，默认化及 wait/direct 运行时待完成 |
|
||||
| ROUTE-007 | 无可用 Upstream 时显式 reject、wait 或 direct，默认 reject | 5075-5294, 6743-6760 | Gateway 已实现 reject、带 `wait_timeout` 的本地容量等待与经 TargetPolicy 的 direct；Distribution 接线和默认化策略待完成 |
|
||||
|
||||
## Provider 与补池
|
||||
|
||||
|
||||
@ -806,6 +806,7 @@ type RoutingRule struct {
|
||||
Upstreams []string `protobuf:"bytes,7,rep,name=upstreams,proto3" json:"upstreams,omitempty"`
|
||||
Strategy *RoutingStrategy `protobuf:"bytes,8,opt,name=strategy,proto3" json:"strategy,omitempty"`
|
||||
OnUnavailable UnavailableAction `protobuf:"varint,9,opt,name=on_unavailable,json=onUnavailable,proto3,enum=proxy_pool.controlplane.v1.UnavailableAction" json:"on_unavailable,omitempty"`
|
||||
WaitTimeout *durationpb.Duration `protobuf:"bytes,10,opt,name=wait_timeout,json=waitTimeout,proto3" json:"wait_timeout,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@ -903,6 +904,13 @@ func (x *RoutingRule) GetOnUnavailable() UnavailableAction {
|
||||
return UnavailableAction_UNAVAILABLE_ACTION_UNSPECIFIED
|
||||
}
|
||||
|
||||
func (x *RoutingRule) GetWaitTimeout() *durationpb.Duration {
|
||||
if x != nil {
|
||||
return x.WaitTimeout
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type RoutingStrategy struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Type StrategyType `protobuf:"varint,1,opt,name=type,proto3,enum=proxy_pool.controlplane.v1.StrategyType" json:"type,omitempty"`
|
||||
@ -2097,7 +2105,7 @@ const file_controlplane_v1_controlplane_proto_rawDesc = "" +
|
||||
"\x10upserted_routing\x18\x06 \x03(\v2'.proxy_pool.controlplane.v1.RoutingRuleR\x0fupsertedRouting\x122\n" +
|
||||
"\x15removed_routing_names\x18\a \x03(\tR\x13removedRoutingNames\x12Q\n" +
|
||||
"\x10upserted_proxies\x18\b \x03(\v2&.proxy_pool.controlplane.v1.OwnedProxyR\x0fupsertedProxies\x12*\n" +
|
||||
"\x11removed_proxy_ids\x18\t \x03(\tR\x0fremovedProxyIds\"\xdc\x03\n" +
|
||||
"\x11removed_proxy_ids\x18\t \x03(\tR\x0fremovedProxyIds\"\x9a\x04\n" +
|
||||
"\vRoutingRule\x12\x12\n" +
|
||||
"\x04name\x18\x01 \x01(\tR\x04name\x12\x18\n" +
|
||||
"\aenabled\x18\x02 \x01(\bR\aenabled\x12\x1d\n" +
|
||||
@ -2109,7 +2117,9 @@ const file_controlplane_v1_controlplane_proto_rawDesc = "" +
|
||||
"\aheaders\x18\x06 \x03(\v24.proxy_pool.controlplane.v1.RoutingRule.HeadersEntryR\aheaders\x12\x1c\n" +
|
||||
"\tupstreams\x18\a \x03(\tR\tupstreams\x12G\n" +
|
||||
"\bstrategy\x18\b \x01(\v2+.proxy_pool.controlplane.v1.RoutingStrategyR\bstrategy\x12T\n" +
|
||||
"\x0eon_unavailable\x18\t \x01(\x0e2-.proxy_pool.controlplane.v1.UnavailableActionR\ronUnavailable\x1a:\n" +
|
||||
"\x0eon_unavailable\x18\t \x01(\x0e2-.proxy_pool.controlplane.v1.UnavailableActionR\ronUnavailable\x12<\n" +
|
||||
"\fwait_timeout\x18\n" +
|
||||
" \x01(\v2\x19.google.protobuf.DurationR\vwaitTimeout\x1a:\n" +
|
||||
"\fHeadersEntry\x12\x10\n" +
|
||||
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
|
||||
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x8a\x02\n" +
|
||||
@ -2337,46 +2347,47 @@ var file_controlplane_v1_controlplane_proto_depIdxs = []int32{
|
||||
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
|
||||
0, // 15: proxy_pool.controlplane.v1.RoutingStrategy.type:type_name -> proxy_pool.controlplane.v1.StrategyType
|
||||
28, // 16: proxy_pool.controlplane.v1.RoutingStrategy.weights:type_name -> proxy_pool.controlplane.v1.RoutingStrategy.WeightsEntry
|
||||
2, // 17: proxy_pool.controlplane.v1.OwnedProxy.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol
|
||||
31, // 18: proxy_pool.controlplane.v1.OwnedProxy.expires_at:type_name -> google.protobuf.Timestamp
|
||||
29, // 19: proxy_pool.controlplane.v1.OwnedProxy.tags:type_name -> proxy_pool.controlplane.v1.OwnedProxy.TagsEntry
|
||||
31, // 20: proxy_pool.controlplane.v1.OwnedProxy.usable_until:type_name -> google.protobuf.Timestamp
|
||||
16, // 21: proxy_pool.controlplane.v1.OutcomeBatch.outcomes:type_name -> proxy_pool.controlplane.v1.ProxyOutcome
|
||||
3, // 22: proxy_pool.controlplane.v1.ProxyOutcome.stage:type_name -> proxy_pool.controlplane.v1.OutcomeStage
|
||||
30, // 23: proxy_pool.controlplane.v1.ProxyOutcome.latency:type_name -> google.protobuf.Duration
|
||||
31, // 24: proxy_pool.controlplane.v1.ProxyOutcome.observed_at:type_name -> google.protobuf.Timestamp
|
||||
19, // 25: proxy_pool.controlplane.v1.ReportRuntimeRequest.counters:type_name -> proxy_pool.controlplane.v1.ProxyRuntime
|
||||
31, // 26: proxy_pool.controlplane.v1.ReportRuntimeRequest.observed_at:type_name -> google.protobuf.Timestamp
|
||||
4, // 27: proxy_pool.controlplane.v1.StreamCheckTasksRequest.supported_levels:type_name -> proxy_pool.controlplane.v1.CheckLevel
|
||||
2, // 28: proxy_pool.controlplane.v1.CheckTask.protocol:type_name -> proxy_pool.controlplane.v1.ProxyProtocol
|
||||
4, // 29: proxy_pool.controlplane.v1.CheckTask.level:type_name -> proxy_pool.controlplane.v1.CheckLevel
|
||||
30, // 30: proxy_pool.controlplane.v1.CheckTask.timeout:type_name -> google.protobuf.Duration
|
||||
31, // 31: proxy_pool.controlplane.v1.CheckTask.deadline:type_name -> google.protobuf.Timestamp
|
||||
24, // 32: proxy_pool.controlplane.v1.ObservationBatch.observations:type_name -> proxy_pool.controlplane.v1.HealthObservation
|
||||
4, // 33: proxy_pool.controlplane.v1.HealthObservation.level:type_name -> proxy_pool.controlplane.v1.CheckLevel
|
||||
30, // 34: proxy_pool.controlplane.v1.HealthObservation.latency:type_name -> google.protobuf.Duration
|
||||
31, // 35: proxy_pool.controlplane.v1.HealthObservation.observed_at:type_name -> google.protobuf.Timestamp
|
||||
5, // 36: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:input_type -> proxy_pool.controlplane.v1.RegisterWorkerRequest
|
||||
7, // 37: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:input_type -> proxy_pool.controlplane.v1.WatchSnapshotsRequest
|
||||
14, // 38: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:input_type -> proxy_pool.controlplane.v1.AcknowledgeSnapshotRequest
|
||||
15, // 39: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:input_type -> proxy_pool.controlplane.v1.OutcomeBatch
|
||||
18, // 40: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:input_type -> proxy_pool.controlplane.v1.ReportRuntimeRequest
|
||||
21, // 41: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:input_type -> proxy_pool.controlplane.v1.StreamCheckTasksRequest
|
||||
23, // 42: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:input_type -> proxy_pool.controlplane.v1.ObservationBatch
|
||||
6, // 43: proxy_pool.controlplane.v1.WorkerControlPlane.RegisterWorker:output_type -> proxy_pool.controlplane.v1.RegisterWorkerResponse
|
||||
8, // 44: proxy_pool.controlplane.v1.WorkerControlPlane.WatchSnapshots:output_type -> proxy_pool.controlplane.v1.SnapshotEnvelope
|
||||
32, // 45: proxy_pool.controlplane.v1.WorkerControlPlane.AcknowledgeSnapshot:output_type -> google.protobuf.Empty
|
||||
17, // 46: proxy_pool.controlplane.v1.WorkerControlPlane.ReportOutcomes:output_type -> proxy_pool.controlplane.v1.ReportOutcomesResponse
|
||||
20, // 47: proxy_pool.controlplane.v1.WorkerControlPlane.ReportRuntime:output_type -> proxy_pool.controlplane.v1.ReportRuntimeResponse
|
||||
22, // 48: proxy_pool.controlplane.v1.CheckerControlPlane.StreamCheckTasks:output_type -> proxy_pool.controlplane.v1.CheckTask
|
||||
25, // 49: proxy_pool.controlplane.v1.CheckerControlPlane.ReportObservations:output_type -> proxy_pool.controlplane.v1.ReportObservationsResponse
|
||||
43, // [43:50] is the sub-list for method output_type
|
||||
36, // [36:43] is the sub-list for method input_type
|
||||
36, // [36:36] is the sub-list for extension type_name
|
||||
36, // [36:36] is the sub-list for extension extendee
|
||||
0, // [0:36] is the sub-list for field type_name
|
||||
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
|
||||
}
|
||||
|
||||
func init() { file_controlplane_v1_controlplane_proto_init() }
|
||||
|
||||
@ -9,6 +9,8 @@ import (
|
||||
controlplanev1 "proxy-pool/gen/controlplane/v1"
|
||||
"proxy-pool/internal/config"
|
||||
"proxy-pool/internal/domain/adminstate"
|
||||
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
)
|
||||
|
||||
var ErrRoutingUnavailable = errors.New("worker snapshot routing is unavailable")
|
||||
@ -110,11 +112,15 @@ func buildGatewayRouting(configuration *config.Config, state *adminstate.Snapsho
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("routing %q unavailable action: %w", configured.Name, err)
|
||||
}
|
||||
var waitTimeout *durationpb.Duration
|
||||
if action == controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT {
|
||||
waitTimeout = durationpb.New(configured.OnUnavailable.WaitTimeout.Value())
|
||||
}
|
||||
rules = append(rules, &controlplanev1.RoutingRule{
|
||||
Name: configured.Name, Enabled: enabled, HostRegex: configured.Match.HostRegex,
|
||||
Methods: append([]string(nil), configured.Match.Methods...), PathRegex: configured.Match.PathRegex,
|
||||
Headers: cloneRoutingHeaders(configured.Match.Headers), Upstreams: upstreams, Strategy: strategy,
|
||||
OnUnavailable: action,
|
||||
OnUnavailable: action, WaitTimeout: waitTimeout,
|
||||
})
|
||||
}
|
||||
return rules, nil
|
||||
|
||||
@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
controlplanev1 "proxy-pool/gen/controlplane/v1"
|
||||
"proxy-pool/internal/config"
|
||||
@ -24,7 +25,7 @@ func TestGatewayRoutingSourceBuildsOrderedRulesFromConfigurationAndAdminState(t
|
||||
{
|
||||
Name: "fallback", Enabled: true, Purpose: "gateway", Upstreams: []string{"provider-a", "provider-c"},
|
||||
Strategy: config.Strategy{Type: "weighted", Weights: map[string]int{"provider-a": 3, "provider-c": 7}},
|
||||
OnUnavailable: config.OnUnavailable{Action: "wait"},
|
||||
OnUnavailable: config.OnUnavailable{Action: "wait", WaitTimeout: config.Duration(25 * time.Millisecond)},
|
||||
},
|
||||
},
|
||||
Upstreams: map[string]config.Upstream{
|
||||
@ -61,7 +62,8 @@ func TestGatewayRoutingSourceBuildsOrderedRulesFromConfigurationAndAdminState(t
|
||||
fallback := rules[1]
|
||||
if !fallback.GetEnabled() || fallback.GetOnUnavailable() != controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT ||
|
||||
!reflect.DeepEqual(fallback.GetUpstreams(), []string{"provider-a", "provider-c"}) ||
|
||||
!reflect.DeepEqual(fallback.GetStrategy().GetWeights(), map[string]uint32{"provider-a": 3, "provider-c": 7}) {
|
||||
!reflect.DeepEqual(fallback.GetStrategy().GetWeights(), map[string]uint32{"provider-a": 3, "provider-c": 7}) ||
|
||||
fallback.GetWaitTimeout().AsDuration() != 25*time.Millisecond {
|
||||
t.Fatalf("fallback rule = %+v", fallback)
|
||||
}
|
||||
}
|
||||
|
||||
@ -5,6 +5,7 @@ import (
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Action string
|
||||
@ -53,6 +54,7 @@ type Rule struct {
|
||||
Action Action
|
||||
Strategy Strategy
|
||||
OnUnavailable OnUnavailableAction
|
||||
WaitTimeout time.Duration
|
||||
}
|
||||
|
||||
type Request struct {
|
||||
|
||||
@ -17,6 +17,7 @@ import (
|
||||
"proxy-pool/internal/gateway/snapshot"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
@ -164,13 +165,18 @@ func wireRouting(source []*controlplanev1.RoutingRule) ([]routingDomain.Rule, er
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
waitTimeout, err := wireWaitTimeout(action, rule.GetWaitTimeout())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result = append(result, routingDomain.Rule{
|
||||
Name: rule.GetName(),
|
||||
Match: routingDomain.Match{
|
||||
HostRegex: rule.GetHostRegex(), Methods: append([]string(nil), rule.GetMethods()...),
|
||||
PathRegex: rule.GetPathRegex(), Headers: cloneRoutingHeaders(rule.GetHeaders()),
|
||||
},
|
||||
Upstreams: upstreams, Action: routingDomain.ActionProxy, Strategy: strategy, OnUnavailable: action,
|
||||
Upstreams: upstreams, Action: routingDomain.ActionProxy, Strategy: strategy,
|
||||
OnUnavailable: action, WaitTimeout: waitTimeout,
|
||||
})
|
||||
}
|
||||
return result, nil
|
||||
@ -246,6 +252,26 @@ func wireUnavailableAction(action controlplanev1.UnavailableAction) (routingDoma
|
||||
}
|
||||
}
|
||||
|
||||
func wireWaitTimeout(action routingDomain.OnUnavailableAction, source *durationpb.Duration) (time.Duration, error) {
|
||||
if source == nil {
|
||||
if action == routingDomain.OnUnavailableWait {
|
||||
return 0, ErrInvalidSnapshotWatcher
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
if err := source.CheckValid(); err != nil {
|
||||
return 0, ErrInvalidSnapshotWatcher
|
||||
}
|
||||
timeout := source.AsDuration()
|
||||
if action == routingDomain.OnUnavailableWait && timeout <= 0 {
|
||||
return 0, ErrInvalidSnapshotWatcher
|
||||
}
|
||||
if action != routingDomain.OnUnavailableWait && timeout != 0 {
|
||||
return 0, ErrInvalidSnapshotWatcher
|
||||
}
|
||||
return timeout, nil
|
||||
}
|
||||
|
||||
func cloneRoutingHeaders(source map[string]string) map[string]string {
|
||||
if source == nil {
|
||||
return nil
|
||||
|
||||
@ -12,6 +12,7 @@ import (
|
||||
"proxy-pool/internal/domain/routing"
|
||||
"proxy-pool/internal/gateway/snapshot"
|
||||
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
@ -27,7 +28,8 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T)
|
||||
Routing: []*controlplanev1.RoutingRule{{
|
||||
Name: "gateway-api", Enabled: true, HostRegex: `^api\.example\.test$`, Upstreams: []string{"upstream-a"},
|
||||
Strategy: &controlplanev1.RoutingStrategy{Type: controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM},
|
||||
OnUnavailable: controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_REJECT,
|
||||
OnUnavailable: controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT,
|
||||
WaitTimeout: durationpb.New(25 * time.Millisecond),
|
||||
}},
|
||||
}
|
||||
setSnapshotChecksum(t, full)
|
||||
@ -45,7 +47,8 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T)
|
||||
t.Fatalf("snapshot view = %+v", view)
|
||||
}
|
||||
matched, ok := view.MatchRouting(routing.Request{Host: "api.example.test", Method: "GET", Path: "/"})
|
||||
if !ok || matched.Name != "gateway-api" || matched.Strategy.Type != routing.StrategyRandom {
|
||||
if !ok || matched.Name != "gateway-api" || matched.Strategy.Type != routing.StrategyRandom ||
|
||||
matched.OnUnavailable != routing.OnUnavailableWait || matched.WaitTimeout != 25*time.Millisecond {
|
||||
t.Fatalf("snapshot routing = %+v, %v", matched, ok)
|
||||
}
|
||||
if client.watch.GetSessionId() != "session-a" || client.ack.GetVersion() != 1 || !client.ack.GetApplied() || string(client.ack.GetChecksum()) != string(full.GetChecksum()) {
|
||||
@ -106,6 +109,30 @@ func TestSnapshotWatcherRejectsExpiredOverallValidityDeadline(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSnapshotWatcherRejectsWaitRoutingWithoutTimeout(t *testing.T) {
|
||||
store := snapshot.NewStore("cluster-a", "worker-a")
|
||||
full := &controlplanev1.WorkerSnapshot{
|
||||
Version: 1, OwnershipEpoch: 7, GeneratedAt: timestamppb.New(time.Now()), ValidUntil: timestamppb.New(time.Now().Add(time.Minute)),
|
||||
Routing: []*controlplanev1.RoutingRule{{
|
||||
Name: "gateway-api", Enabled: true, HostRegex: `^api\.example\.test$`, Upstreams: []string{"upstream-a"},
|
||||
Strategy: &controlplanev1.RoutingStrategy{Type: controlplanev1.StrategyType_STRATEGY_TYPE_RANDOM},
|
||||
OnUnavailable: controlplanev1.UnavailableAction_UNAVAILABLE_ACTION_WAIT,
|
||||
}},
|
||||
}
|
||||
setSnapshotChecksum(t, full)
|
||||
client := &snapshotClientStub{stream: &snapshotStreamStub{values: []*controlplanev1.SnapshotEnvelope{{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: full}}}}}
|
||||
watcher, err := NewSnapshotWatcher(client, store, SnapshotWatcherOptions{ClusterID: "cluster-a", WorkerID: "worker-a"})
|
||||
if err != nil {
|
||||
t.Fatalf("NewSnapshotWatcher(): %v", err)
|
||||
}
|
||||
if err := watcher.Watch(context.Background(), "session-a"); err == nil {
|
||||
t.Fatal("Watch() error = nil, want invalid wait routing rejection")
|
||||
}
|
||||
if client.ack == nil || client.ack.GetApplied() || client.ack.GetErrorCode() != "snapshot_apply_failed" {
|
||||
t.Fatalf("negative acknowledgement = %+v", client.ack)
|
||||
}
|
||||
}
|
||||
|
||||
type snapshotClientStub struct {
|
||||
stream SnapshotStream
|
||||
watch *controlplanev1.WatchSnapshotsRequest
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
package dispatch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@ -17,6 +18,7 @@ type Request struct {
|
||||
RoutingName string
|
||||
Strategy routing.Strategy
|
||||
OnUnavailable routing.OnUnavailableAction
|
||||
WaitTimeout time.Duration
|
||||
Now time.Time
|
||||
Scheme proxyDomain.Scheme
|
||||
Upstreams []string
|
||||
@ -74,6 +76,32 @@ func (d *Dispatcher) Acquire(request Request) (*Lease, error) {
|
||||
return d.acquireFromUpstreams(view, request, request.Upstreams)
|
||||
}
|
||||
|
||||
// AcquireWait retries local snapshot dispatch at a bounded interval until a
|
||||
// capacity slot appears, the route timeout expires, or the caller cancels.
|
||||
func (d *Dispatcher) AcquireWait(ctx context.Context, request Request, timeout time.Duration) (*Lease, error) {
|
||||
if d == nil || ctx == nil || timeout <= 0 {
|
||||
return nil, ErrNoCandidate
|
||||
}
|
||||
deadline := time.NewTimer(timeout)
|
||||
defer deadline.Stop()
|
||||
retry := time.NewTicker(5 * time.Millisecond)
|
||||
defer retry.Stop()
|
||||
for {
|
||||
request.Now = time.Now().UTC()
|
||||
lease, err := d.Acquire(request)
|
||||
if !errors.Is(err, ErrNoCandidate) {
|
||||
return lease, err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-deadline.C:
|
||||
return nil, ErrNoCandidate
|
||||
case <-retry.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Dispatcher) acquireRouted(view *snapshot.View, request Request) (*Lease, error) {
|
||||
if request.Strategy.Type == routing.StrategySequential {
|
||||
if request.Strategy.CurrentUpstream == "" {
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
package dispatch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
@ -292,6 +293,35 @@ func TestAcquireUsesLeastConnectionsAndSkipsFullUpstream(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAcquireWaitsForReleasedLocalCapacity(t *testing.T) {
|
||||
store := routingStrategyStore(t)
|
||||
dispatcher := New(store)
|
||||
held, err := dispatcher.Acquire(Request{Now: time.Now(), Upstreams: []string{"upstream-a"}})
|
||||
if err != nil {
|
||||
t.Fatalf("Acquire(held): %v", err)
|
||||
}
|
||||
if err := held.Commit(); err != nil {
|
||||
t.Fatalf("Commit(held): %v", err)
|
||||
}
|
||||
released := make(chan struct{})
|
||||
go func() {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
_ = held.Release()
|
||||
close(released)
|
||||
}()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
lease, err := dispatcher.AcquireWait(ctx, Request{Upstreams: []string{"upstream-a"}}, 100*time.Millisecond)
|
||||
if err != nil {
|
||||
t.Fatalf("AcquireWait(): %v", err)
|
||||
}
|
||||
if err := lease.Cancel(); err != nil {
|
||||
t.Fatalf("Cancel(): %v", err)
|
||||
}
|
||||
<-released
|
||||
}
|
||||
|
||||
func TestAcquireSurvivesConcurrentSnapshotApply(t *testing.T) {
|
||||
now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC)
|
||||
store := snapshot.NewStore("cluster-a", "worker-a")
|
||||
|
||||
@ -15,6 +15,7 @@ import (
|
||||
"time"
|
||||
|
||||
proxyDomain "proxy-pool/internal/domain/proxy"
|
||||
"proxy-pool/internal/domain/routing"
|
||||
"proxy-pool/internal/gateway/dispatch"
|
||||
"proxy-pool/internal/gateway/policy"
|
||||
transportDomain "proxy-pool/internal/gateway/transport"
|
||||
@ -72,6 +73,15 @@ type ProxyTransport interface {
|
||||
Relay(context.Context, net.Conn, net.Conn) error
|
||||
}
|
||||
|
||||
type DirectTransport interface {
|
||||
RoundTripDirect(context.Context, *http.Request) (*http.Response, error)
|
||||
OpenDirectTunnel(context.Context, string) (net.Conn, error)
|
||||
}
|
||||
|
||||
type waitingDispatcher interface {
|
||||
AcquireWait(context.Context, dispatch.Request, time.Duration) (*dispatch.Lease, error)
|
||||
}
|
||||
|
||||
type Dependencies struct {
|
||||
Auth Guard
|
||||
Access Guard
|
||||
@ -206,9 +216,23 @@ func (handler *Handler) connect(
|
||||
route.Now = time.Now().UTC()
|
||||
route.Exclude = excluded
|
||||
route.SafetyMargin = handler.config.SafetyMargin
|
||||
lease, err := handler.dispatcher.Acquire(route)
|
||||
lease, err := handler.acquireRoute(request.Context(), route)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
if errors.Is(err, dispatch.ErrNoCandidate) && route.OnUnavailable == routing.OnUnavailableDirect {
|
||||
direct, directErr := handler.directTransport()
|
||||
if directErr != nil {
|
||||
lastErr = directErr
|
||||
break
|
||||
}
|
||||
upstream, directErr := direct.OpenDirectTunnel(request.Context(), target.DialAddress())
|
||||
if directErr != nil {
|
||||
lastErr = directErr
|
||||
break
|
||||
}
|
||||
handler.serveTunnel(writer, request, nil, upstream)
|
||||
return
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
@ -235,7 +259,7 @@ func (handler *Handler) connect(
|
||||
lastErr = err
|
||||
break
|
||||
}
|
||||
handler.serveTunnel(writer, request, lease, upstream)
|
||||
handler.serveTunnel(writer, request, func() { finishLease(lease, true) }, upstream)
|
||||
return
|
||||
}
|
||||
|
||||
@ -253,10 +277,12 @@ func (handler *Handler) writeConnectError(writer http.ResponseWriter, responseEr
|
||||
func (handler *Handler) serveTunnel(
|
||||
writer http.ResponseWriter,
|
||||
request *http.Request,
|
||||
lease *dispatch.Lease,
|
||||
finish func(),
|
||||
upstream net.Conn,
|
||||
) {
|
||||
defer finishLease(lease, true)
|
||||
if finish != nil {
|
||||
defer finish()
|
||||
}
|
||||
defer upstream.Close()
|
||||
|
||||
hijacker, ok := writer.(http.Hijacker)
|
||||
@ -390,9 +416,23 @@ func (handler *Handler) forwardHTTP(
|
||||
route.Now = time.Now().UTC()
|
||||
route.Exclude = excluded
|
||||
route.SafetyMargin = handler.config.SafetyMargin
|
||||
lease, err := handler.dispatcher.Acquire(route)
|
||||
lease, err := handler.acquireRoute(request.Context(), route)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
if errors.Is(err, dispatch.ErrNoCandidate) && route.OnUnavailable == routing.OnUnavailableDirect {
|
||||
direct, directErr := handler.directTransport()
|
||||
if directErr != nil {
|
||||
lastErr = directErr
|
||||
break
|
||||
}
|
||||
response, directErr := direct.RoundTripDirect(request.Context(), attemptRequest)
|
||||
if directErr != nil {
|
||||
lastErr = directErr
|
||||
break
|
||||
}
|
||||
handler.writeResponse(writer, response)
|
||||
return
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
@ -427,6 +467,26 @@ func (handler *Handler) forwardHTTP(
|
||||
writeGatewayError(writer, fmt.Errorf("forward gateway request: %w", lastErr))
|
||||
}
|
||||
|
||||
func (handler *Handler) acquireRoute(ctx context.Context, route dispatch.Request) (*dispatch.Lease, error) {
|
||||
lease, err := handler.dispatcher.Acquire(route)
|
||||
if !errors.Is(err, dispatch.ErrNoCandidate) || route.OnUnavailable != routing.OnUnavailableWait || route.WaitTimeout <= 0 {
|
||||
return lease, err
|
||||
}
|
||||
waiter, ok := handler.dispatcher.(waitingDispatcher)
|
||||
if !ok {
|
||||
return nil, err
|
||||
}
|
||||
return waiter.AcquireWait(ctx, route, route.WaitTimeout)
|
||||
}
|
||||
|
||||
func (handler *Handler) directTransport() (DirectTransport, error) {
|
||||
direct, ok := handler.transport.(DirectTransport)
|
||||
if !ok {
|
||||
return nil, ErrDirectRouteUnsupported
|
||||
}
|
||||
return direct, nil
|
||||
}
|
||||
|
||||
func pinHTTPDestination(request *http.Request, target policy.Authority) {
|
||||
if request == nil || request.URL == nil || (!target.ResolvedIP.IsValid() && !target.LiteralIP.IsValid()) {
|
||||
return
|
||||
|
||||
@ -16,6 +16,7 @@ import (
|
||||
"time"
|
||||
|
||||
proxyDomain "proxy-pool/internal/domain/proxy"
|
||||
"proxy-pool/internal/domain/routing"
|
||||
"proxy-pool/internal/gateway/dispatch"
|
||||
"proxy-pool/internal/gateway/policy"
|
||||
"proxy-pool/internal/gateway/snapshot"
|
||||
@ -207,6 +208,47 @@ func TestHandlerPinsValidatedHTTPDestinationWithoutChangingHost(t *testing.T) {
|
||||
assertNoLeakedCapacity(t, view)
|
||||
}
|
||||
|
||||
func TestHandlerUsesDirectFallbackAfterTargetPolicy(t *testing.T) {
|
||||
var directCalls int
|
||||
transport := &fakeTransport{directRoundTrip: func(_ context.Context, request *http.Request) (*http.Response, error) {
|
||||
directCalls++
|
||||
if request.URL.Host != "198.51.100.10:80" || request.Host != "example.test" {
|
||||
t.Fatalf("direct request = URL %q Host %q", request.URL.Host, request.Host)
|
||||
}
|
||||
return &http.Response{StatusCode: http.StatusNoContent, Header: make(http.Header), Body: http.NoBody}, nil
|
||||
}}
|
||||
handler, err := New(Config{}, Dependencies{
|
||||
Targets: fakeTargets{evaluateURL: func(context.Context, string) (policy.Authority, error) {
|
||||
return policy.Authority{Host: "example.test", Port: 80, ResolvedIP: netip.MustParseAddr("198.51.100.10")}, nil
|
||||
}},
|
||||
Router: RouteFunc(func(*http.Request) (dispatch.Request, error) {
|
||||
return dispatch.Request{OnUnavailable: routing.OnUnavailableDirect}, nil
|
||||
}),
|
||||
Dispatcher: DispatcherFunc(func(dispatch.Request) (*dispatch.Lease, error) { return nil, dispatch.ErrNoCandidate }),
|
||||
Transport: transport,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("New(): %v", err)
|
||||
}
|
||||
|
||||
response := httptest.NewRecorder()
|
||||
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "http://example.test/resource", nil))
|
||||
if response.Code != http.StatusNoContent || directCalls != 1 {
|
||||
t.Fatalf("response = %d, direct calls = %d", response.Code, directCalls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerWaitsOnlyForWaitRoutingAction(t *testing.T) {
|
||||
dispatcher := &waitRecordingDispatcher{}
|
||||
handler := &Handler{dispatcher: dispatcher}
|
||||
_, err := handler.acquireRoute(context.Background(), dispatch.Request{
|
||||
OnUnavailable: routing.OnUnavailableWait, WaitTimeout: 25 * time.Millisecond,
|
||||
})
|
||||
if !errors.Is(err, dispatch.ErrNoCandidate) || dispatcher.waitTimeout != 25*time.Millisecond {
|
||||
t.Fatalf("acquireRoute() error = %v, wait timeout = %s", err, dispatcher.waitTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerEnforcesConcurrentRequestLimit(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@ -449,9 +491,36 @@ type fakeTransport struct {
|
||||
seen []string
|
||||
roundTrip func(context.Context, proxyDomain.Proxy, *http.Request, ...func() error) (*http.Response, error)
|
||||
openTunnel func(context.Context, proxyDomain.Proxy, string) (net.Conn, error)
|
||||
directRoundTrip func(context.Context, *http.Request) (*http.Response, error)
|
||||
directTunnel func(context.Context, string) (net.Conn, error)
|
||||
relay func(context.Context, net.Conn, net.Conn) error
|
||||
}
|
||||
|
||||
type waitRecordingDispatcher struct{ waitTimeout time.Duration }
|
||||
|
||||
func (dispatcher *waitRecordingDispatcher) Acquire(dispatch.Request) (*dispatch.Lease, error) {
|
||||
return nil, dispatch.ErrNoCandidate
|
||||
}
|
||||
|
||||
func (dispatcher *waitRecordingDispatcher) AcquireWait(_ context.Context, _ dispatch.Request, timeout time.Duration) (*dispatch.Lease, error) {
|
||||
dispatcher.waitTimeout = timeout
|
||||
return nil, dispatch.ErrNoCandidate
|
||||
}
|
||||
|
||||
func (transport *fakeTransport) RoundTripDirect(ctx context.Context, request *http.Request) (*http.Response, error) {
|
||||
if transport.directRoundTrip == nil {
|
||||
return nil, errors.New("direct round trip not configured")
|
||||
}
|
||||
return transport.directRoundTrip(ctx, request)
|
||||
}
|
||||
|
||||
func (transport *fakeTransport) OpenDirectTunnel(ctx context.Context, target string) (net.Conn, error) {
|
||||
if transport.directTunnel == nil {
|
||||
return nil, errors.New("direct tunnel not configured")
|
||||
}
|
||||
return transport.directTunnel(ctx, target)
|
||||
}
|
||||
|
||||
func (transport *fakeTransport) RoundTrip(
|
||||
ctx context.Context,
|
||||
selected proxyDomain.Proxy,
|
||||
|
||||
@ -89,7 +89,7 @@ func routeWithMatcher(request *http.Request, match routeMatcher) (dispatch.Reque
|
||||
case routing.ActionProxy:
|
||||
return dispatch.Request{
|
||||
RoutingName: matched.Name, Upstreams: append([]string(nil), matched.Upstreams...),
|
||||
Strategy: matched.Strategy, OnUnavailable: matched.OnUnavailable,
|
||||
Strategy: matched.Strategy, OnUnavailable: matched.OnUnavailable, WaitTimeout: matched.WaitTimeout,
|
||||
}, nil
|
||||
case routing.ActionReject:
|
||||
return dispatch.Request{}, fmt.Errorf("%w: %s", ErrRouteRejected, matched.Name)
|
||||
|
||||
@ -66,7 +66,7 @@ func TestSnapshotRouterMatchesRulesFromCurrentSnapshot(t *testing.T) {
|
||||
Name: "gateway-api", Match: routing.Match{HostRegex: `^api\.example\.test$`},
|
||||
Upstreams: []string{"provider-a"}, Action: routing.ActionProxy,
|
||||
Strategy: routing.Strategy{Type: routing.StrategyWeighted, Weights: map[string]uint32{"provider-a": 3}},
|
||||
OnUnavailable: routing.OnUnavailableReject,
|
||||
OnUnavailable: routing.OnUnavailableWait, WaitTimeout: 25 * time.Millisecond,
|
||||
}},
|
||||
}
|
||||
envelope.Checksum = snapshot.ChecksumWithRouting(envelope.Proxies, envelope.Routing)
|
||||
@ -80,7 +80,8 @@ func TestSnapshotRouterMatchesRulesFromCurrentSnapshot(t *testing.T) {
|
||||
t.Fatalf("Route(): %v", err)
|
||||
}
|
||||
if result.RoutingName != "gateway-api" || result.Strategy.Type != routing.StrategyWeighted ||
|
||||
!reflect.DeepEqual(result.Strategy.Weights, map[string]uint32{"provider-a": 3}) {
|
||||
!reflect.DeepEqual(result.Strategy.Weights, map[string]uint32{"provider-a": 3}) ||
|
||||
result.OnUnavailable != routing.OnUnavailableWait || result.WaitTimeout != 25*time.Millisecond {
|
||||
t.Fatalf("Route() = %+v", result)
|
||||
}
|
||||
}
|
||||
|
||||
@ -446,6 +446,7 @@ func cloneRoutingRules(source []routing.Rule) []routing.Rule {
|
||||
result := make([]routing.Rule, len(source))
|
||||
for index, rule := range source {
|
||||
result[index] = rule
|
||||
result[index].WaitTimeout = rule.WaitTimeout
|
||||
result[index].Match.Methods = append([]string(nil), rule.Match.Methods...)
|
||||
result[index].Upstreams = append([]string(nil), rule.Upstreams...)
|
||||
if rule.Match.Headers != nil {
|
||||
|
||||
@ -88,24 +88,34 @@ type Transport struct {
|
||||
config Config
|
||||
resolver CredentialResolver
|
||||
client *http.Transport
|
||||
direct *http.Transport
|
||||
buffers sync.Pool
|
||||
}
|
||||
|
||||
func New(config Config, resolver CredentialResolver) *Transport {
|
||||
applyDefaults(&config)
|
||||
transport := &Transport{config: config, resolver: resolver}
|
||||
tlsConfig := config.TLSClientConfig
|
||||
if tlsConfig != nil {
|
||||
tlsConfig = tlsConfig.Clone()
|
||||
}
|
||||
transport.client = &http.Transport{
|
||||
Proxy: func(request *http.Request) (*url.URL, error) {
|
||||
transport.client = newHTTPTransport(config, func(request *http.Request) (*url.URL, error) {
|
||||
proxyURL, ok := request.Context().Value(proxyURLContextKey{}).(*url.URL)
|
||||
if !ok || proxyURL == nil {
|
||||
return nil, errors.New("upstream proxy URL is missing from request context")
|
||||
}
|
||||
return proxyURL, nil
|
||||
},
|
||||
})
|
||||
transport.direct = newHTTPTransport(config, nil)
|
||||
transport.buffers.New = func() any {
|
||||
return make([]byte, config.TunnelBufferBytes)
|
||||
}
|
||||
return transport
|
||||
}
|
||||
|
||||
func newHTTPTransport(config Config, proxy func(*http.Request) (*url.URL, error)) *http.Transport {
|
||||
tlsConfig := config.TLSClientConfig
|
||||
if tlsConfig != nil {
|
||||
tlsConfig = tlsConfig.Clone()
|
||||
}
|
||||
return &http.Transport{
|
||||
Proxy: proxy,
|
||||
DialContext: (&net.Dialer{Timeout: config.DialTimeout, KeepAlive: 30 * time.Second}).DialContext,
|
||||
ForceAttemptHTTP2: true,
|
||||
MaxIdleConns: config.MaxIdleConns,
|
||||
@ -116,10 +126,6 @@ func New(config Config, resolver CredentialResolver) *Transport {
|
||||
TLSClientConfig: tlsConfig,
|
||||
MaxResponseHeaderBytes: config.MaxResponseHeaderBytes,
|
||||
}
|
||||
transport.buffers.New = func() any {
|
||||
return make([]byte, config.TunnelBufferBytes)
|
||||
}
|
||||
return transport
|
||||
}
|
||||
|
||||
func (transport *Transport) RoundTrip(
|
||||
@ -170,6 +176,30 @@ func (transport *Transport) RoundTrip(
|
||||
return response, err
|
||||
}
|
||||
|
||||
func (transport *Transport) RoundTripDirect(ctx context.Context, request *http.Request) (*http.Response, error) {
|
||||
if transport == nil || transport.direct == nil || request == nil {
|
||||
return nil, errors.New("direct round trip: nil transport or request")
|
||||
}
|
||||
clone := request.Clone(ctx)
|
||||
clone.RequestURI = ""
|
||||
clone.Header = request.Header.Clone()
|
||||
clone.Header.Del("Proxy-Authorization")
|
||||
return transport.direct.RoundTrip(clone)
|
||||
}
|
||||
|
||||
func (transport *Transport) OpenDirectTunnel(ctx context.Context, target string) (net.Conn, error) {
|
||||
if transport == nil || target == "" {
|
||||
return nil, errors.New("open direct tunnel: target is required")
|
||||
}
|
||||
dialContext, cancel := context.WithTimeout(ctx, transport.config.DialTimeout)
|
||||
defer cancel()
|
||||
connection, err := (&net.Dialer{Timeout: transport.config.DialTimeout, KeepAlive: 30 * time.Second}).DialContext(dialContext, "tcp", target)
|
||||
if err != nil {
|
||||
return nil, contextError(ctx, fmt.Errorf("dial direct target %s: %w", target, err))
|
||||
}
|
||||
return connection, nil
|
||||
}
|
||||
|
||||
func (transport *Transport) OpenTunnel(
|
||||
ctx context.Context,
|
||||
selected proxyDomain.Proxy,
|
||||
@ -256,6 +286,9 @@ func (transport *Transport) CloseIdleConnections() {
|
||||
return
|
||||
}
|
||||
transport.client.CloseIdleConnections()
|
||||
if transport.direct != nil {
|
||||
transport.direct.CloseIdleConnections()
|
||||
}
|
||||
}
|
||||
|
||||
func (transport *Transport) Relay(ctx context.Context, left, right net.Conn) error {
|
||||
|
||||
@ -65,6 +65,58 @@ func TestRoundTripForwardsHTTPViaSelectedProxy(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoundTripDirectForwardsWithoutProxyAuthorization(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
upstream := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
||||
if request.Header.Get("Proxy-Authorization") != "" {
|
||||
t.Fatalf("Proxy-Authorization = %q, want empty", request.Header.Get("Proxy-Authorization"))
|
||||
}
|
||||
writer.WriteHeader(http.StatusNoContent)
|
||||
}))
|
||||
defer upstream.Close()
|
||||
|
||||
request := httptest.NewRequest(http.MethodGet, upstream.URL+"/resource", nil)
|
||||
request.Header.Set("Proxy-Authorization", "Basic should-not-forward")
|
||||
response, err := New(Config{}, nil).RoundTripDirect(request.Context(), request)
|
||||
if err != nil {
|
||||
t.Fatalf("RoundTripDirect(): %v", err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != http.StatusNoContent {
|
||||
t.Fatalf("status = %d, want 204", response.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenDirectTunnelDialsTarget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("Listen(): %v", err)
|
||||
}
|
||||
defer listener.Close()
|
||||
accepted := make(chan struct{})
|
||||
go func() {
|
||||
connection, acceptErr := listener.Accept()
|
||||
if acceptErr == nil {
|
||||
_ = connection.Close()
|
||||
close(accepted)
|
||||
}
|
||||
}()
|
||||
|
||||
connection, err := New(Config{DialTimeout: time.Second}, nil).OpenDirectTunnel(context.Background(), listener.Addr().String())
|
||||
if err != nil {
|
||||
t.Fatalf("OpenDirectTunnel(): %v", err)
|
||||
}
|
||||
defer connection.Close()
|
||||
select {
|
||||
case <-accepted:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("target listener did not accept direct tunnel")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoundTripCommitsReservationAfterConnectionAcquisition(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user