From 265026a9071b57d421c65df5935d0222363fef9a Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 15:55:50 +0800 Subject: [PATCH] feat: handle unavailable gateway routes --- README.md | 7 +- api/proto/controlplane/v1/controlplane.proto | 1 + docs/api/control-plane.md | 6 +- docs/development/implementation-plan.md | 6 +- docs/requirements/completion-audit.md | 6 +- docs/requirements/traceability.md | 2 +- gen/controlplane/v1/controlplane.pb.go | 95 +++++++++++-------- .../worker/routing_snapshot_source.go | 8 +- .../worker/routing_snapshot_source_test.go | 6 +- internal/domain/routing/rule.go | 2 + internal/gateway/controlplane/watcher.go | 28 +++++- internal/gateway/controlplane/watcher_test.go | 31 +++++- internal/gateway/dispatch/dispatcher.go | 28 ++++++ internal/gateway/dispatch/dispatcher_test.go | 30 ++++++ internal/gateway/server/handler.go | 70 +++++++++++++- internal/gateway/server/handler_test.go | 79 ++++++++++++++- internal/gateway/server/routing.go | 2 +- internal/gateway/server/routing_test.go | 5 +- internal/gateway/snapshot/store.go | 1 + internal/gateway/transport/transport.go | 57 ++++++++--- internal/gateway/transport/transport_test.go | 52 ++++++++++ 21 files changed, 437 insertions(+), 85 deletions(-) diff --git a/README.md b/README.md index f4cdf22..5480952 100644 --- a/README.md +++ b/README.md @@ -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` 仍只是待验证的集群设计目标。 diff --git a/api/proto/controlplane/v1/controlplane.proto b/api/proto/controlplane/v1/controlplane.proto index c19ce48..5474d43 100644 --- a/api/proto/controlplane/v1/controlplane.proto +++ b/api/proto/controlplane/v1/controlplane.proto @@ -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 { diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 2fda535..429b642 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -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 和已签发引用，因此迟到的 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index ca422b2..2030837 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -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 和分配安全余量的批次；已覆盖 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index bded8fe..2bd444f 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -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 中回收。 diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 355def3..3f5608a 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -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 与补池 diff --git a/gen/controlplane/v1/controlplane.pb.go b/gen/controlplane/v1/controlplane.pb.go index 315ec5d..405f2ca 100644 --- a/gen/controlplane/v1/controlplane.pb.go +++ b/gen/controlplane/v1/controlplane.pb.go @@ -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() } diff --git a/internal/controller/worker/routing_snapshot_source.go b/internal/controller/worker/routing_snapshot_source.go index 9d8f959..28279ad 100644 --- a/internal/controller/worker/routing_snapshot_source.go +++ b/internal/controller/worker/routing_snapshot_source.go @@ -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 diff --git a/internal/controller/worker/routing_snapshot_source_test.go b/internal/controller/worker/routing_snapshot_source_test.go index cc2c297..048eb9d 100644 --- a/internal/controller/worker/routing_snapshot_source_test.go +++ b/internal/controller/worker/routing_snapshot_source_test.go @@ -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) } } diff --git a/internal/domain/routing/rule.go b/internal/domain/routing/rule.go index 4eae180..4b6163e 100644 --- a/internal/domain/routing/rule.go +++ b/internal/domain/routing/rule.go @@ -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 { diff --git a/internal/gateway/controlplane/watcher.go b/internal/gateway/controlplane/watcher.go index bde721a..967e93c 100644 --- a/internal/gateway/controlplane/watcher.go +++ b/internal/gateway/controlplane/watcher.go @@ -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 diff --git a/internal/gateway/controlplane/watcher_test.go b/internal/gateway/controlplane/watcher_test.go index b11aed0..220fdbb 100644 --- a/internal/gateway/controlplane/watcher_test.go +++ b/internal/gateway/controlplane/watcher_test.go @@ -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 diff --git a/internal/gateway/dispatch/dispatcher.go b/internal/gateway/dispatch/dispatcher.go index ed99295..d22d2af 100644 --- a/internal/gateway/dispatch/dispatcher.go +++ b/internal/gateway/dispatch/dispatcher.go @@ -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 == "" { diff --git a/internal/gateway/dispatch/dispatcher_test.go b/internal/gateway/dispatch/dispatcher_test.go index 89070c6..86c182e 100644 --- a/internal/gateway/dispatch/dispatcher_test.go +++ b/internal/gateway/dispatch/dispatcher_test.go @@ -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") diff --git a/internal/gateway/server/handler.go b/internal/gateway/server/handler.go index a49dd00..a2fb329 100644 --- a/internal/gateway/server/handler.go +++ b/internal/gateway/server/handler.go @@ -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 diff --git a/internal/gateway/server/handler_test.go b/internal/gateway/server/handler_test.go index 09ee653..886eda9 100644 --- a/internal/gateway/server/handler_test.go +++ b/internal/gateway/server/handler_test.go @@ -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() @@ -445,11 +487,38 @@ func (targets fakeTargets) EvaluateConnectAuthority(ctx context.Context, raw str } type fakeTransport struct { - mu sync.Mutex - 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) - relay func(context.Context, net.Conn, net.Conn) error + mu sync.Mutex + 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( diff --git a/internal/gateway/server/routing.go b/internal/gateway/server/routing.go index 980f6eb..a985704 100644 --- a/internal/gateway/server/routing.go +++ b/internal/gateway/server/routing.go @@ -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) diff --git a/internal/gateway/server/routing_test.go b/internal/gateway/server/routing_test.go index 759ec5b..693ffd7 100644 --- a/internal/gateway/server/routing_test.go +++ b/internal/gateway/server/routing_test.go @@ -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) } } diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index 3ec89d1..6fb2df7 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -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 { diff --git a/internal/gateway/transport/transport.go b/internal/gateway/transport/transport.go index 0426e98..425c5e5 100644 --- a/internal/gateway/transport/transport.go +++ b/internal/gateway/transport/transport.go @@ -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} + 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() } - transport.client = &http.Transport{ - Proxy: 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 - }, + 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 { diff --git a/internal/gateway/transport/transport_test.go b/internal/gateway/transport/transport_test.go index fdfc69a..ad729f3 100644 --- a/internal/gateway/transport/transport_test.go +++ b/internal/gateway/transport/transport_test.go @@ -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()