From 2166214777d5f6b2d563e13e48fe02d635629d04 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 09:16:46 +0800 Subject: [PATCH] feat: schedule routing target health checks --- README.md | 23 ++-- docs/api/control-plane.md | 14 ++- docs/configuration/reference.md | 14 ++- docs/development/implementation-plan.md | 24 ++-- docs/operations/runbook.md | 6 +- docs/requirements/completion-audit.md | 5 +- docs/requirements/traceability.md | 6 +- .../adapters/redisactivity/health_tasks.go | 82 ++++++++++-- .../health_tasks_integration_test.go | 60 +++++++++ internal/adapters/redisactivity/keys.go | 2 + internal/adapters/redisactivity/scripts.go | 13 +- .../redisactivity/scripts/health_tasks.lua | 79 ++++++++---- internal/config/config.go | 12 ++ internal/config/redact.go | 1 + internal/config/routing_check_test.go | 80 ++++++++++++ internal/config/validate.go | 50 ++++++-- .../controller/health/scheduler_runner.go | 119 ++++++++++++++++-- .../health/scheduler_runner_test.go | 109 ++++++++++++++++ 18 files changed, 599 insertions(+), 100 deletions(-) create mode 100644 internal/config/routing_check_test.go diff --git a/README.md b/README.md index c87b6f7..7d0505d 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,7 @@ flowchart LR ## 当前完成度 -截至 **2026-08-02**，实施计划中可直接勾选的检查项为 **53 / 74（71.6%）**。详情见 +截至 **2026-08-02**，实施计划中可直接勾选的检查项为 **57 / 74（77.0%）**。详情见 [实施计划](docs/development/implementation-plan.md)和 [交付完成度审计](docs/requirements/completion-audit.md)。 @@ -88,11 +88,10 @@ flowchart LR 提取与限流、Controller 的 Admin/Distribution/Metrics 监听，以及 PostgreSQL 管理状态；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和 Redis 会话栅栏，以及 Gateway Outcome 上报的有界队列、序列确认与重试； - Controller 的 Redis 共享 BASIC/EGRESS 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 + Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 Checker 探测和 Observation 状态归并。 -- **部分完成**：TARGET 的任务编排和配置建模，Docker Compose/Kubernetes - 运行时 mTLS Overlay。 +- **部分完成**：Docker Compose/Kubernetes 运行时 mTLS Overlay。 - **待完成**：CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 @@ -169,11 +168,12 @@ Checker 的参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、 `PROXY_POOL_CHECKER_ID`、`PROXY_POOL_CHECKER_INSTANCE_ID` 与 `PROXY_POOL_CHECKER_MAX_IN_FLIGHT` 提供。它不会访问 Redis/PostgreSQL；生产 Controller 在启用控制面时装配 Redis 共享任务 broker，并按启用的 Upstream 调度 -HTTP/HTTPS/SOCKS5 BASIC 检查，并按每个 `check.urls` 创建 EGRESS 任务。调度监督器每轮读取 -已发布配置，因此 reload 后的上游启停、检查间隔、抖动、超时、重试次数、`maxInFlight` 和 -EGRESS URL 都会在下一轮生效；BASIC 与 EGRESS 以有界轮转组共享上游并发上限。新启用的上游 -无需重启 Controller。EGRESS 对成功响应提取纯文本 IP 或常见 JSON IP 字段，并将其作为全局 -健康事实回传；TARGET 尚未进入生产调度。 +HTTP/HTTPS/SOCKS5 BASIC 检查、按每个 `check.urls` 创建 EGRESS 任务，并按启用 Routing 的 +`check.targets` 创建 TARGET 任务。调度监督器每轮读取已发布配置，因此 reload 后的上游/路由启停、 +检查间隔、抖动、超时、重试次数、`maxInFlight`、EGRESS URL 和 TARGET Profile 都会在下一轮生效； +BASIC、EGRESS 与 TARGET 以有界轮转组共享上游并发上限。新启用的上游无需重启 Controller。 +EGRESS 对成功响应提取纯文本 IP 或常见 JSON IP 字段，并将其作为全局健康事实回传；TARGET 事实 +仅归并到对应的 `(routing_name, target_url)` Profile，不改变 Proxy 全局健康。 初版 HTTP 容量工具可按固定请求数或固定时长运行，并将 HTTPS 目标经 Gateway 的请求 交给标准 HTTP Transport 建立 CONNECT： @@ -229,8 +229,9 @@ go run ./cmd/proxy-loadgen ` - **P0 - Worker 控制面闭环**：Worker session、Snapshot ledger、ACK、运行态接收、 ownership 索引，以及 Gateway 快照客户端。 -- **P0 - Checker 健康链**：BASIC/EGRESS 的共享调度、实际探测、Observation reducer 和 - `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成；继续补齐 TARGET。 +- **P0 - Checker 健康链**：BASIC/EGRESS/TARGET 的共享调度、实际探测、Observation reducer 和 + `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成；TARGET 事实按路由目标 Profile + 独立归并。 - **P1 - Gateway 与 Routing**：Gateway 进程、快照凭据分发、五种 Routing 策略与 `onUnavailable` 已接入；动态容量调整和 Drain 闭环待完成。 - **P1 - 可观测与部署**：低基数业务指标、完整 Compose/Kubernetes 进程拓扑， diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index 1002830..b06d2f4 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -43,7 +43,9 @@ Redis 只保留每个当前会话的一条序列和摘要；原始 Outcome、代 Redis 或 PostgreSQL。Gateway 只将结果写入本地有界队列，队列满时丢弃样本，不等待控制面 或存储。Checker 的有界任务领取、任务租约归属校验和 Observation 上报已经由 gRPC 契约测试覆盖；Controller 已装配 Redis 共享 due-index、生产任务 broker 与独立 -Checker 的 HTTP/HTTPS/SOCKS5 BASIC 探测进程。EGRESS 和 TARGET 尚未进入生产调度。 +Checker 的 HTTP/HTTPS/SOCKS5 BASIC、EGRESS、TARGET 探测进程。TARGET Profile 按 +`(routing_name, target_url)` 独立调度和归并，BASIC/EGRESS/TARGET 共享每个 Upstream 的 +in-flight 上限。 `100,000 QPS` 仍是未验证的设计目标。 `WatchSnapshots` 建立时校验当前 session；每次签发快照引用时也把 `session_id` @@ -174,9 +176,9 @@ Routing 决定 AVAILABLE、SUSPECT 或 UNHEALTHY，并更新 Redis 活动池， 相同领取者对已确认任务的同一事实可重放，由活动池摘要幂等处理。EGRESS 的探测 URL 只存在于下发任务，Checker 回传时不携带 URL 或 Routing Profile，因此它只归并 Proxy 全局健康。生产 Controller 装配 Redis 共享 broker，按启用 Upstream 的有效检查策略 -调度 BASIC 任务；未装配 broker 的 fixture 服务仍会以 `Unavailable` 拒绝任务流， -而不下发无租约任务。EGRESS 和 TARGET 的 -任务索引及调度策略尚未实现。 +调度 BASIC/EGRESS 任务，并按启用 Routing 的 `check.targets` 调度 TARGET 任务；未装配 +broker 的 fixture 服务仍会以 `Unavailable` 拒绝任务流，而不下发无租约任务。三类任务 +使用独立 due-index 和引用，且共享 Upstream in-flight 上限。 每次 Claim 还会签发新的不可预测 `lease_token`；Observation 必须回传该值。任务被重新 领取后，旧 token 即使拥有相同 `task_id` 和 `checker_id` 也会被拒绝，避免过期实例的 迟到事实覆盖新租约结果。 @@ -185,8 +187,8 @@ Routing 决定 AVAILABLE、SUSPECT 或 UNHEALTHY，并更新 Redis 活动池， `max_in_flight`；每次尝试都受 `deadline` 和 `timeout` 的较小值约束，失败可在同一 deadline 内最多执行到 `max_attempts`。BASIC 针对 HTTP/HTTPS/SOCKS5 Proxy 验证到 Proxy 的 请求/认证握手；EGRESS 与 TARGET 通过 Proxy 请求任务指定的 HTTP/HTTPS 目标并将 -非成功状态作为事实。EGRESS 对成功响应解析纯文本或常见 JSON IP 字段；EGRESS 生产调度 -已完成，TARGET 生产调度仍待后续实现。 +非成功状态作为事实。EGRESS 对成功响应解析纯文本或常见 JSON IP 字段；EGRESS 与 TARGET +均已进入生产调度。 ## 7. 兼容与演进 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index eec0579..95a5a38 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -202,9 +202,9 @@ Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-pla `PROXY_POOL_*` 环境变量提供。mTLS 模式下该命令读取 `checkerTLS`，明文 fixture 模式只接受回环 Controller 地址。Checker 只从 gRPC 领取任务并批量上报事实，不读取 Redis/PostgreSQL；Controller 在生产启动拓扑中装配 Redis 共享任务队列，当前调度 -HTTP/HTTPS/SOCKS5 BASIC 检查。调度监督器在每轮从已发布配置读取启用的上游；Admin reload -发布后，上游启停和有效 `check` 策略会在下一轮生效，新启用的上游无需重启 Controller。 -EGRESS 和 TARGET 的生产调度仍在后续实施范围。 +HTTP/HTTPS/SOCKS5 BASIC、EGRESS 和 TARGET 检查。调度监督器在每轮从已发布配置读取启用的 +上游与 Routing；Admin reload 发布后，上游/路由启停、有效 `check` 策略和目标列表会在下一轮生效， +新启用的上游无需重启 Controller。 `maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway 在本地维护容量为 `65536` 的非阻塞 Outcome 队列，默认微批上限为 `512`，实际取二者中 @@ -304,6 +304,8 @@ routing: onUnavailable: action: reject waitTimeout: 0s + check: + targets: [https://api.example.com/health] ``` - Routing 列表有序，首条匹配后停止。 @@ -315,6 +317,9 @@ routing: `switchAfterEmptyFetch`；`endBehavior` 省略时默认为 `stop`，也可显式设置 `loop` 或 `stayLast`。 - `onUnavailable.action` 为 `reject`、`wait` 或 `direct`；默认建议 `reject`。 +- `check.targets` 是 Routing 级 HTTP/HTTPS 探测目标。每个 Routing 最多 16 个 URL，且一个 + 启用 Upstream 被其引用的 TARGET Profile 总数最多 64 个。每个 Profile 的身份是 + `(routing.name, target URL)`；失败只影响该 Profile，不改变 Proxy 的全局健康状态。 Sequential 的空计数属于 Upstream，当前索引属于 Routing。只有 Provider 响应 成功、模板成功且合法候选为零时才增加空计数。错误不改变空计数；重复候选 @@ -467,6 +472,9 @@ proxyAuth: - `check.urls` 最多 16 个规范化 HTTP/HTTPS URL，作为 EGRESS 检查的有界目标集； BASIC 检查不依赖该字段。EGRESS 成功响应必须包含合法 IP：支持纯文本 IP，或 JSON 的 `ip`、`ip_address`、`address`、`origin`、`query` 字段。 +- 启用 Routing 的 `check.targets` 会为其引用的每个启用 Upstream 创建 TARGET 检查组。BASIC、 + EGRESS 与 TARGET 使用固定批次和轮转顺序，并共享该 Upstream 的 `check.maxInFlight`，避免 + 配置多个目标后产生无界检查流量。 - 第一次有意义失败进入 SUSPECT；达到 `maxConsecutiveFailures` 后才进入 UNHEALTHY。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index c814bf8..b36aa8f 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -261,17 +261,17 @@ Controller 多副本共享同一计数。Client 身份只以 SHA-256 摘要进 Gateway 请求热路径仍只使用本地准入，不增加 Redis/PostgreSQL 调用。 WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本、基础 Snapshot 流、Gateway 会话客户端与快照凭据分发已完成；权威 Proxy/Routing 发布、 -Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重试；健康 BASIC/EGRESS 执行链已 -完成，TARGET 编排和 REMOVE 生命周期仍待完成，因此 Task 10 尚未全部完成。 +Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重试；健康 BASIC/EGRESS/TARGET +执行链已完成，REMOVE 生命周期仍待完成，因此 Task 10 尚未全部完成。 ## Task 11: Checker and Health Reducer **Files:** `internal/controller/health/*.go`, `cmd/proxy-checker/main.go`, tests -- [ ] Schedule global and route health with jitter and bounded maxInFlight. -- [ ] Implement FETCHED -> CHECKING -> AVAILABLE and SUSPECT/UNHEALTHY transitions. -- [ ] Ensure target failures affect only the target profile. -- [ ] Add fixture target server and deterministic clock/scheduler tests. +- [x] Schedule global and route health with jitter and bounded maxInFlight. +- [x] Implement FETCHED -> CHECKING -> AVAILABLE and SUSPECT/UNHEALTHY transitions. +- [x] Ensure target failures affect only the target profile. +- [x] Add fixture target server and deterministic clock/scheduler tests. 当前进度（2026-07-31）：已新增不依赖 gRPC 的 `domain/health` Observation、全局/Target Profile Reducer 与任务摘要幂等语义；BASIC/EGRESS 结果经活动池窄接口在 Memory 和 Redis @@ -286,14 +286,16 @@ Redis 共享 due-index/租约持久化、每 Upstream 的跨副本 in-flight 限 `proxy-checker` 独立进程、固定大小 worker-pool、任务期重试/微批上报和 HTTP/HTTPS/SOCKS5 BASIC/EGRESS 探测器已完成并有测试。Redis 已按固定 EGRESS due-index 保存任务执行 URL， 与 BASIC 独立引用并通过上游共享并发限制；配置化监督器以有界轮转组调度每个 `check.urls`。 -TARGET 探测器具备任务执行能力，但尚无生产任务调度；EGRESS 的出口身份响应解析已支持固定 -上限的纯文本和常见 JSON IP 字段，部署运行态仍未实现， -因此本任务保持未完成。 +EGRESS 的出口身份响应解析支持固定上限的纯文本和常见 JSON IP 字段。TARGET 已以 +`routing.check.targets` 作为配置入口，限定每个 Routing 的 URL 数和每个 Upstream 的 Profile +总数；Redis 为 `(routing_name, target_url, proxy_id)` 维护独立 due-index，完成、租约回收和 +代理过期均沿用同一原子任务边界。BASIC、EGRESS 和 TARGET 分组轮转并共享 Upstream in-flight +上限，因此本任务已完成。 补充进度（2026-08-02）：BASIC 调度已改为配置驱动监督器。它每轮读取已发布快照并复用 有界派发逻辑，所以 reload 后已启用上游的策略变更、停用，以及新启用上游都无需重启 -Controller 即可生效。Redis 任务存储现已扩展 BASIC/EGRESS 的独立有界索引；TARGET 的多维 -索引仍待实现。 +Controller 即可生效。Redis 任务存储现已扩展 BASIC/EGRESS/TARGET 的独立有界索引；路由目标 +Profile 在启用 Routing 与 Upstream 的组合上才进入调度。 补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 场景。固定请求数和固定时长两种 模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小延迟直方图， diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index d8998d9..bb963ca 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -23,9 +23,9 @@ `cmd/proxy-controller` 已完成配置单次加载、PostgreSQL 迁移、Redis 活动池、 Distribution/Admin/Metrics 独立监听和有界停机装配。Provider 自动补池、分布式 -配额、动态重载和 Admin 低基数统计已装配；Controller 已装配 Redis BASIC/EGRESS 任务 broker， -`proxy-checker` 可执行 HTTP/HTTPS/SOCKS5 BASIC/EGRESS 探测。`proxy-loadgen` 已提供有界 HTTP -请求场景；TARGET 调度、CONNECT 长连接/Extract 压测与完整 mTLS 环境 Overlay 仍属于 +配额、动态重载和 Admin 低基数统计已装配；Controller 已装配 Redis BASIC/EGRESS/TARGET 任务 broker， +`proxy-checker` 可执行 HTTP/HTTPS/SOCKS5 BASIC/EGRESS/TARGET 探测。`proxy-loadgen` 已提供有界 HTTP +请求场景；CONNECT 长连接/Extract 压测与完整 mTLS 环境 Overlay 仍属于 `implementation-plan.md` 后续任务。 因此 Compose/Kubernetes 资产当前仍用于评审网络、资源、探针和依赖关系，不能 视为完整可运行拓扑。 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 45d3b71..91c2738 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -62,8 +62,9 @@ Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controll 记录归并，不改写 Proxy 全局状态。Controller 公用 Reducer 已作为 Observation 的唯一状态 归并边界，Checker Observation RPC 已在同一控制面监听以独立 SPIFFE 身份接入；有界任务领取、 租约归属与任务期凭据传输已由通用契约和 gRPC 往返测试覆盖。Redis 共享 due-index、 -任务/租约持久化、按上游的 in-flight 限制以及独立 Checker 的 HTTP/HTTPS/SOCKS5 BASIC/EGRESS 执行 -进程已经闭环；EGRESS 已具备多维任务索引、出口身份响应解析与生产调度，TARGET 的多维索引仍未实现。Snapshot 签发在 Redis 中原子匹配当前 +任务/租约持久化、按上游的 in-flight 限制以及独立 Checker 的 HTTP/HTTPS/SOCKS5 BASIC/EGRESS/TARGET 执行 +进程已经闭环；EGRESS 已具备多维任务索引、出口身份响应解析与生产调度；TARGET 已按 +`(routing_name, target_url)` 的独立有界索引和调度组进入生产链。Snapshot 签发在 Redis 中原子匹配当前 `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流；Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index b296966..59aab01 100644 --- a/docs/requirements/traceability.md +++ b/docs/requirements/traceability.md @@ -7,7 +7,7 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS 共享任务运行态、`proxy-loadgen` 有界 HTTP 场景均已实现；TARGET 生产编排待实现 | +| ARCH-001 | 数据面 Worker 与控制面 Controller 分离 | 1-70 | 包、协议和部署拓扑已分离；Controller 已运行 Worker Register/Watch/ACK/Runtime/Outcome 与 Checker Observation gRPC，并发布 Proxy/Gateway Routing/按引用去重凭据完整快照；Checker 任务流已具备有界领取、租约栅栏和任务期凭据契约。Gateway 已将快照编译为同版本动态 View，并由独立进程维护控制面会话。`proxy-checker` 与 Redis BASIC/EGRESS/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP 场景均已实现 | | ARCH-002 | 热路径只做认证、本地路由和网络转发 | 1-70, 380-430 | Gateway bootstrap 集成测试验证启动期控制面会话与快照就绪，HTTP 请求只走本地 Snapshot/Dispatch；Outcome 仅写入有界非阻塞本地队列，代表性性能剖析待完成 | | ARCH-003 | Gateway、Distribution、Admin、Metrics 独立入口 | 8904-8958 | Controller 命令已装配 Distribution/Admin/Metrics 三个独立监听及联动停机；Gateway 命令已装配代理与 Metrics 监听，运行时 mTLS 部署 Overlay 待完成 | | ARCH-004 | Controller 集中 Provider 获取与切换 | 1403-1580 | Redis Leader、动态 Provider Supervisor 与 Bootstrap 生产装配已完成；Admin disable/reload 驱动取消替换，多副本按权威 HMAC 指纹和 revision 栅栏收敛并拒绝旧配置换主；Routing 切换到 Drain 的编排待完成 | @@ -79,8 +79,8 @@ | ID | 最终需求 | 来源 | 验证证据 | |---|---|---|---| -| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 在 Memory 和 Redis 独立、随代理 TTL 归并，不改写 Proxy 全局状态；Routing 消费待实现 | -| HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置有效合并、URL 校验、稳定抖动/优先级 Planner、有界 Scheduler tick、Redis BASIC/EGRESS due-index、跨副本 in-flight 原子限制，以及 HTTP/HTTPS/SOCKS5 BASIC/EGRESS 生产执行器已完成；TARGET 多维任务索引与调度待实现 | +| HEALTH-001 | 全局健康与 Routing/目标健康分离 | 221-270, 8679-8708 | `domain/health` 已将 BASIC/EGRESS 全局 Reducer 与 TARGET Profile Reducer 分离；TARGET 在 Memory 和 Redis 独立、随代理 TTL 归并，不改写 Proxy 全局状态，并按 `(routing_name, target_url)` 进入生产调度 | +| HEALTH-002 | 健康调度有 jitter、maxInFlight 和分级频率 | 8679-8736 | 配置有效合并、URL/Profile 上限、稳定抖动/优先级 Planner、有界 Scheduler tick、Redis BASIC/EGRESS/TARGET due-index、跨副本 in-flight 原子限制，以及 HTTP/HTTPS/SOCKS5 BASIC/EGRESS/TARGET 生产执行器已完成 | | HEALTH-003 | 失败分级 SUSPECT -> UNHEALTHY -> REMOVE | 8679-8736 | Controller 公用 Reducer 已通过 Memory/Redis 活动池原子提交全局连续失败、精确重放和成功恢复；BASIC 任务调度已完成，REMOVE 编排待实现 | | SEC-001 | API 认证与 Proxy 认证分离，Secret 统一脱敏 | 7528-8111, 8904-8945 | Config 脱敏、Provider Store -> SecretRef -> Gateway Resolver 跨包测试与格式化泄漏回归测试 | | SEC-002 | 非回环监听无保护时严格模式启动失败 | 8112-8441 | 配置校验测试 | diff --git a/internal/adapters/redisactivity/health_tasks.go b/internal/adapters/redisactivity/health_tasks.go index 88d53ed..2358185 100644 --- a/internal/adapters/redisactivity/health_tasks.go +++ b/internal/adapters/redisactivity/health_tasks.go @@ -52,6 +52,7 @@ type healthTaskRequest struct { Levels []healthDomain.Level `json:"levels,omitempty"` UpstreamID string `json:"upstreamId,omitempty"` Level string `json:"level,omitempty"` + RoutingName string `json:"routingName,omitempty"` TargetURL string `json:"targetUrl,omitempty"` UpstreamTasksKey string `json:"upstreamTasksKey,omitempty"` ScanLimit int `json:"scanLimit,omitempty"` @@ -88,14 +89,14 @@ func (a *Adapter) InFlight(ctx context.Context, now time.Time) (int, error) { // DueCandidates returns a bounded BASIC due batch. Stale references are // discarded inside the Lua script before they can reach the Controller. func (a *Adapter) DueCandidates(ctx context.Context, now time.Time, limit int) ([]healthDomain.Candidate, error) { - return a.dueCandidates(ctx, "", healthDomain.LevelBasic, "", now, limit) + return a.dueCandidates(ctx, "", healthDomain.LevelBasic, "", "", now, limit) } func (a *Adapter) DueCandidatesForUpstream(ctx context.Context, upstreamID string, now time.Time, limit int) ([]healthDomain.Candidate, error) { if !validHealthTaskIdentifier(upstreamID) { return nil, healthDomain.ErrInvalidTaskBroker } - return a.dueCandidates(ctx, upstreamID, healthDomain.LevelBasic, "", now, limit) + return a.dueCandidates(ctx, upstreamID, healthDomain.LevelBasic, "", "", now, limit) } // DueEgressCandidatesForUpstream lazily initializes bounded EGRESS references @@ -109,18 +110,50 @@ func (a *Adapter) DueEgressCandidatesForUpstream(ctx context.Context, upstreamID if err != nil || normalizedTarget != targetURL { return nil, healthDomain.ErrInvalidTaskBroker } - return a.dueCandidates(ctx, upstreamID, healthDomain.LevelEgress, targetURL, now, limit) + return a.dueCandidates(ctx, upstreamID, healthDomain.LevelEgress, targetURL, "", now, limit) } -func (a *Adapter) dueCandidates(ctx context.Context, upstreamID string, level healthDomain.Level, targetURL string, now time.Time, limit int) ([]healthDomain.Candidate, error) { +// DueTargetCandidatesForUpstream lazily initializes bounded TARGET references +// from live BASIC due items. TARGET health remains isolated by the configured +// Routing name and URL, so one target failure cannot change global proxy state. +func (a *Adapter) DueTargetCandidatesForUpstream( + ctx context.Context, + upstreamID string, + routingName string, + targetURL string, + now time.Time, + limit int, +) ([]healthDomain.Candidate, error) { + if !validHealthTaskIdentifier(upstreamID) { + return nil, healthDomain.ErrInvalidTaskBroker + } + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: routingName, TargetURL: targetURL, + }) + if err != nil || profile.RoutingName != routingName || profile.TargetURL != targetURL { + return nil, healthDomain.ErrInvalidTaskBroker + } + return a.dueCandidates(ctx, upstreamID, healthDomain.LevelTarget, targetURL, routingName, now, limit) +} + +func (a *Adapter) dueCandidates( + ctx context.Context, + upstreamID string, + level healthDomain.Level, + targetURL string, + routingName string, + now time.Time, + limit int, +) ([]healthDomain.Candidate, error) { if a == nil || ctx == nil || now.IsZero() || limit <= 0 || limit > a.options.MaxCheckTasks { return nil, healthDomain.ErrInvalidTaskBroker } - if level != healthDomain.LevelBasic && level != healthDomain.LevelEgress { + if level != healthDomain.LevelBasic && level != healthDomain.LevelEgress && level != healthDomain.LevelTarget { return nil, healthDomain.ErrInvalidTaskBroker } reply, err := a.runHealthTaskScript(ctx, "due", now, healthTaskRequest{ - Limit: limit, ScanLimit: a.options.MaxCandidateScan, UpstreamID: upstreamID, Level: string(level), TargetURL: targetURL, + Limit: limit, ScanLimit: a.options.MaxCandidateScan, UpstreamID: upstreamID, Level: string(level), + RoutingName: routingName, TargetURL: targetURL, }, "runtime") if err != nil { return nil, err @@ -144,14 +177,23 @@ func (a *Adapter) dueCandidates(ctx context.Context, upstreamID string, level he healthDomain.Level(candidate.Level) != level { return nil, invalidScriptReply("health task due reply contains invalid candidate") } - if level == healthDomain.LevelEgress { + switch level { + case healthDomain.LevelEgress: normalizedTarget, err := healthDomain.NormalizeEgressTarget(candidate.TargetURL) if err != nil || normalizedTarget != targetURL { return nil, invalidScriptReply("health task due reply contains invalid EGRESS target") } + case healthDomain.LevelTarget: + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: candidate.RoutingName, TargetURL: candidate.TargetURL, + }) + if err != nil || profile.RoutingName != routingName || profile.TargetURL != targetURL { + return nil, invalidScriptReply("health task due reply contains invalid TARGET profile") + } } result = append(result, healthDomain.Candidate{ - ProxyID: candidate.ProxyID, UpstreamID: candidate.UpstreamID, State: state, Level: level, TargetURL: candidate.TargetURL, + ProxyID: candidate.ProxyID, UpstreamID: candidate.UpstreamID, State: state, Level: level, + RoutingName: candidate.RoutingName, TargetURL: candidate.TargetURL, DueAt: time.UnixMilli(candidate.DueAtMS).UTC(), }) } @@ -352,7 +394,7 @@ func (a *Adapter) runHealthTaskScript( result, err := runScript(ctx, a.client, healthTasksScript, []string{ a.keys.records, a.keys.expiry, a.keys.healthDue, a.keys.healthQueued, a.keys.healthLeases, a.keys.healthTasks, a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.checkerLeases(checkerID), - a.keys.stateInventory, a.keys.healthEgressDue, + a.keys.stateInventory, a.keys.healthEgressDue, a.keys.healthTargetDue, }, operation, now.UTC().UnixMilli(), string(payload)) if err != nil { return healthTaskScriptReply{}, err @@ -380,6 +422,13 @@ func validateRedisPlannedTask(task healthDomain.PlannedTask, now time.Time) erro if task.Candidate.RoutingName != "" || err != nil || normalizedTarget != task.Candidate.TargetURL { return healthDomain.ErrInvalidLeasedTask } + case healthDomain.LevelTarget: + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: task.Candidate.RoutingName, TargetURL: task.Candidate.TargetURL, + }) + if err != nil || profile.RoutingName != task.Candidate.RoutingName || profile.TargetURL != task.Candidate.TargetURL { + return healthDomain.ErrInvalidLeasedTask + } default: return healthDomain.ErrInvalidLeasedTask } @@ -443,10 +492,16 @@ func redisHealthTaskReference(candidate healthDomain.Candidate) string { } func redisHealthDueMember(candidate healthDomain.Candidate) string { - if candidate.Level == healthDomain.LevelBasic { + switch candidate.Level { + case healthDomain.LevelBasic: return candidate.ProxyID + case healthDomain.LevelEgress: + return candidate.TargetURL + "\x00" + candidate.ProxyID + case healthDomain.LevelTarget: + return candidate.RoutingName + "\x00" + candidate.TargetURL + "\x00" + candidate.ProxyID + default: + return "" } - return candidate.TargetURL + "\x00" + candidate.ProxyID } func validRedisTaskExecution(task healthTaskRecord) bool { @@ -456,6 +511,11 @@ func validRedisTaskExecution(task healthTaskRecord) bool { case healthDomain.LevelEgress: targetURL, err := healthDomain.NormalizeEgressTarget(task.TargetURL) return task.RoutingName == "" && err == nil && targetURL == task.TargetURL + case healthDomain.LevelTarget: + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: task.RoutingName, TargetURL: task.TargetURL, + }) + return err == nil && profile.RoutingName == task.RoutingName && profile.TargetURL == task.TargetURL default: return false } diff --git a/internal/adapters/redisactivity/health_tasks_integration_test.go b/internal/adapters/redisactivity/health_tasks_integration_test.go index 0d7c185..39572b7 100644 --- a/internal/adapters/redisactivity/health_tasks_integration_test.go +++ b/internal/adapters/redisactivity/health_tasks_integration_test.go @@ -179,3 +179,63 @@ func TestRedisHealthTasksLeaseAndRescheduleEgressChecks(t *testing.T) { t.Fatalf("DueEgressCandidatesForUpstream(next) = (%+v, %v)", next, err) } } + +func TestRedisHealthTasksLeaseAndRescheduleTargetChecks(t *testing.T) { + fixture := newRedisTestFixture(t) + ctx := context.Background() + now := time.Now().UTC() + const routingName = "checkout" + const targetURL = "https://checkout.example/health" + if _, err := fixture.Adapter.UpsertFetched(ctx, "provider-a", activitypool.FetchedBatch{ + ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1, + Proxies: []proxyDomain.Proxy{testProxy("proxy-a", "192.0.2.10")}, + }); err != nil { + t.Fatalf("UpsertFetched(): %v", err) + } + candidates, err := fixture.Adapter.DueTargetCandidatesForUpstream(ctx, "provider-a", routingName, targetURL, now, 1) + if err != nil || len(candidates) != 1 || candidates[0].Level != healthDomain.LevelTarget || + candidates[0].RoutingName != routingName || candidates[0].TargetURL != targetURL { + t.Fatalf("DueTargetCandidatesForUpstream() = (%+v, %v)", candidates, err) + } + planner, err := controllerHealth.NewPlanner(controllerHealth.SchedulePolicy{ + Interval: 30 * time.Second, MaxInFlight: 2, Timeout: 5 * time.Second, MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("NewPlanner(): %v", err) + } + plans, err := planner.Plan(now, 0, 1, candidates) + if err != nil || len(plans) != 1 { + t.Fatalf("Plan() = (%+v, %v)", plans, err) + } + if offered, err := fixture.Adapter.Offer(ctx, plans); err != nil || offered != 1 { + t.Fatalf("Offer() = (%d, %v)", offered, err) + } + claimed, err := fixture.Adapter.Claim(ctx, healthDomain.TaskClaim{ + CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 1, + SupportedLevels: []healthDomain.Level{healthDomain.LevelTarget}, + }) + if err != nil || len(claimed) != 1 || claimed[0].Level != healthDomain.LevelTarget || + claimed[0].RoutingName != routingName || claimed[0].TargetURL != targetURL { + t.Fatalf("Claim() = (%+v, %v)", claimed, err) + } + observation := healthDomain.Observation{ + TaskID: claimed[0].TaskID, ProxyID: claimed[0].ProxyID, Level: healthDomain.LevelTarget, + RoutingName: routingName, TargetURL: targetURL, Success: true, Latency: time.Millisecond, ObservedAt: time.Now().UTC(), + } + if err := fixture.Adapter.AuthorizeObservation(ctx, "checker-a", claimed[0].LeaseToken, observation, time.Now().UTC()); err != nil { + t.Fatalf("AuthorizeObservation() = %v", err) + } + state, err := fixture.Adapter.ApplyTargetObservation(ctx, activitypool.TargetHealthCommand{ + Observation: observation, MaxConsecutiveFailures: 2, + }) + if err != nil || state.Status != healthDomain.TargetAvailable { + t.Fatalf("ApplyTargetObservation() = (%+v, %v)", state, err) + } + if err := fixture.Adapter.CompleteObservation(ctx, "checker-a", claimed[0].LeaseToken, observation, time.Now().UTC()); err != nil { + t.Fatalf("CompleteObservation() = %v", err) + } + next, err := fixture.Adapter.DueTargetCandidatesForUpstream(ctx, "provider-a", routingName, targetURL, plans[0].NextDue, 1) + if err != nil || len(next) != 1 || next[0].ProxyID != "proxy-a" || next[0].RoutingName != routingName || next[0].TargetURL != targetURL { + t.Fatalf("DueTargetCandidatesForUpstream(next) = (%+v, %v)", next, err) + } +} diff --git a/internal/adapters/redisactivity/keys.go b/internal/adapters/redisactivity/keys.go index 46eacd3..93a9481 100644 --- a/internal/adapters/redisactivity/keys.go +++ b/internal/adapters/redisactivity/keys.go @@ -29,6 +29,7 @@ type keyspace struct { workerOutcomes string healthDue string healthEgressDue string + healthTargetDue string healthQueued string healthLeases string healthTasks string @@ -59,6 +60,7 @@ func newKeyspace(namespace string) keyspace { workerOutcomes: prefix + ":worker-outcomes", healthDue: prefix + ":health-due", healthEgressDue: prefix + ":health-egress-due", + healthTargetDue: prefix + ":health-target-due", healthQueued: prefix + ":health-queued", healthLeases: prefix + ":health-leases", healthTasks: prefix + ":health-tasks", diff --git a/internal/adapters/redisactivity/scripts.go b/internal/adapters/redisactivity/scripts.go index 378b33f..c8f0006 100644 --- a/internal/adapters/redisactivity/scripts.go +++ b/internal/adapters/redisactivity/scripts.go @@ -52,12 +52,13 @@ type healthTaskScriptReply struct { } type healthTaskCandidateWire struct { - ProxyID string `json:"proxyId"` - UpstreamID string `json:"upstreamId"` - State string `json:"state"` - Level string `json:"level"` - TargetURL string `json:"targetUrl,omitempty"` - DueAtMS int64 `json:"dueAtMs"` + ProxyID string `json:"proxyId"` + UpstreamID string `json:"upstreamId"` + State string `json:"state"` + Level string `json:"level"` + RoutingName string `json:"routingName,omitempty"` + TargetURL string `json:"targetUrl,omitempty"` + DueAtMS int64 `json:"dueAtMs"` } type healthTaskClaimWire struct { diff --git a/internal/adapters/redisactivity/scripts/health_tasks.lua b/internal/adapters/redisactivity/scripts/health_tasks.lua index 0f0a454..41a1b1c 100644 --- a/internal/adapters/redisactivity/scripts/health_tasks.lua +++ b/internal/adapters/redisactivity/scripts/health_tasks.lua @@ -9,6 +9,7 @@ local ref_task_key = KEYS[8] local checker_leases_key = KEYS[9] local state_inventory_key = KEYS[10] local egress_due_key = KEYS[11] +local target_due_key = KEYS[12] local operation = ARGV[1] local now_ms = tonumber(ARGV[2]) @@ -48,6 +49,26 @@ local function touch(key, expires_at_ms) end end +local function due_key_for_level(level) + if level == 'EGRESS' then + return egress_due_key + end + if level == 'TARGET' then + return target_due_key + end + return due_key +end + +local function task_reference(level, proxy_id, routing_name, target_url) + if level == 'EGRESS' then + return proxy_id .. '\0EGRESS\0\0' .. target_url + end + if level == 'TARGET' then + return proxy_id .. '\0TARGET\0' .. routing_name .. '\0' .. target_url + end + return proxy_id +end + local function remove_task(task_id, requeue) local raw = redis.call('HGET', tasks_key, task_id) local task = nil @@ -77,10 +98,7 @@ local function remove_task(task_id, requeue) if requeue and type(task.proxyId) == 'string' and type(due_member) == 'string' then local _, record = live_record(task.proxyId) if record then - local task_due_key = due_key - if task.level == 'EGRESS' then - task_due_key = egress_due_key - end + local task_due_key = due_key_for_level(task.level) redis.call('ZADD', task_due_key, now_ms, due_member) touch(task_due_key, tonumber(record.expiresAtMs)) end @@ -211,12 +229,16 @@ if operation == 'due' then return finish({status = 'invalid'}) end local level = payload.level or 'BASIC' - if level ~= 'BASIC' and level ~= 'EGRESS' then + if level ~= 'BASIC' and level ~= 'EGRESS' and level ~= 'TARGET' then return finish({status = 'invalid'}) end if level == 'EGRESS' and (type(payload.targetUrl) ~= 'string' or payload.targetUrl == '') then return finish({status = 'invalid'}) end + if level == 'TARGET' and (type(payload.routingName) ~= 'string' or payload.routingName == '' or + type(payload.targetUrl) ~= 'string' or payload.targetUrl == '') then + return finish({status = 'invalid'}) + end local result_members = {} local function append_candidate(proxy_id, record, member) if result_members[member] then @@ -225,28 +247,30 @@ if operation == 'due' then result_members[member] = true result[#result + 1] = { proxyId = proxy_id, upstreamId = record.sourceUpstream, state = record.state, level = level, - targetUrl = level == 'EGRESS' and payload.targetUrl or '', dueAtMs = now_ms + routingName = level == 'TARGET' and payload.routingName or '', + targetUrl = (level == 'EGRESS' or level == 'TARGET') and payload.targetUrl or '', dueAtMs = now_ms } end - if level == 'EGRESS' then + if level == 'EGRESS' or level == 'TARGET' then local basic_ids = redis.call('ZRANGEBYSCORE', due_key, '-inf', now_ms, 'LIMIT', 0, scan_limit) for _, proxy_id in ipairs(basic_ids) do local _, record = live_record(proxy_id) if record and valid_state(record.state) and (type(payload.upstreamId) ~= 'string' or payload.upstreamId == '' or record.sourceUpstream == payload.upstreamId) then local member = payload.targetUrl .. '\0' .. proxy_id - local reference = proxy_id .. '\0EGRESS\0\0' .. payload.targetUrl + if level == 'TARGET' then + member = payload.routingName .. '\0' .. payload.targetUrl .. '\0' .. proxy_id + end + local reference = task_reference(level, proxy_id, payload.routingName or '', payload.targetUrl) if not redis.call('HGET', ref_task_key, reference) then - redis.call('ZADD', egress_due_key, now_ms, member) - touch(egress_due_key, tonumber(record.expiresAtMs)) + local task_due_key = due_key_for_level(level) + redis.call('ZADD', task_due_key, now_ms, member) + touch(task_due_key, tonumber(record.expiresAtMs)) end end end end - local source_due_key = due_key - if level == 'EGRESS' then - source_due_key = egress_due_key - end + local source_due_key = due_key_for_level(level) local ids = redis.call('ZRANGEBYSCORE', source_due_key, '-inf', now_ms, 'LIMIT', 0, scan_limit) for _, member in ipairs(ids) do local proxy_id = member @@ -258,13 +282,22 @@ if operation == 'due' then else proxy_id = string.sub(member, separator + 1) end + elseif level == 'TARGET' then + local first_separator = string.find(member, '\0', 1, true) + local second_separator = first_separator and string.find(member, '\0', first_separator + 1, true) or nil + if not first_separator or not second_separator or string.sub(member, 1, first_separator - 1) ~= payload.routingName or + string.sub(member, first_separator + 1, second_separator - 1) ~= payload.targetUrl then + matching_target = false + else + proxy_id = string.sub(member, second_separator + 1) + end end if matching_target then local _, record = live_record(proxy_id) - local reference = proxy_id .. '\0EGRESS\0\0' .. payload.targetUrl + local reference = task_reference(level, proxy_id, payload.routingName or '', payload.targetUrl or '') if not record or not valid_state(record.state) then redis.call('ZREM', source_due_key, member) - elseif level == 'EGRESS' and redis.call('HGET', ref_task_key, reference) then + elseif (level == 'EGRESS' or level == 'TARGET') and redis.call('HGET', ref_task_key, reference) then redis.call('ZREM', source_due_key, member) elseif type(payload.upstreamId) == 'string' and payload.upstreamId ~= '' and record.sourceUpstream ~= payload.upstreamId then -- The ref remains due for the scheduler that owns this upstream. @@ -289,7 +322,7 @@ if operation == 'offer' then end local offered = 0 for _, task in ipairs(payload.tasks) do - if type(task) ~= 'table' or task.version ~= 1 or (task.level ~= 'BASIC' and task.level ~= 'EGRESS') or task.state ~= 'QUEUED' or + if type(task) ~= 'table' or task.version ~= 1 or (task.level ~= 'BASIC' and task.level ~= 'EGRESS' and task.level ~= 'TARGET') or task.state ~= 'QUEUED' or type(task.taskId) ~= 'string' or type(task.proxyId) ~= 'string' or tonumber(task.deadlineMs or 0) <= now_ms or type(task.upstreamId) ~= 'string' or task.upstreamId == '' or type(task.upstreamTasksKey) ~= 'string' or task.upstreamTasksKey == '' or type(task.reference) ~= 'string' or task.reference == '' or @@ -303,10 +336,11 @@ if operation == 'offer' then if task.level == 'EGRESS' and ((task.routingName or '') ~= '' or type(task.targetUrl) ~= 'string' or task.targetUrl == '') then return finish({status = 'invalid'}) end - local task_due_key = due_key - if task.level == 'EGRESS' then - task_due_key = egress_due_key + if task.level == 'TARGET' and (type(task.routingName) ~= 'string' or task.routingName == '' or + type(task.targetUrl) ~= 'string' or task.targetUrl == '') then + return finish({status = 'invalid'}) end + local task_due_key = due_key_for_level(task.level) local score = redis.call('ZSCORE', task_due_key, task.dueMember) local current = redis.call('HGET', ref_task_key, task.reference) local raw, record = live_record(task.proxyId) @@ -427,10 +461,7 @@ if operation == 'authorize' or operation == 'complete' then if next_due_ms <= now_ms then next_due_ms = now_ms end - local task_due_key = due_key - if task.level == 'EGRESS' then - task_due_key = egress_due_key - end + local task_due_key = due_key_for_level(task.level) redis.call('ZADD', task_due_key, next_due_ms, task.dueMember or task.proxyId) touch(task_due_key, tonumber(record.expiresAtMs)) end diff --git a/internal/config/config.go b/internal/config/config.go index f709dba..6d0c4b6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -9,6 +9,11 @@ import ( // scheduling and proxy cleanup proportional to a fixed configuration limit. const MaximumCheckURLs = 16 +// MaximumTargetProfilesPerUpstream bounds the Cartesian expansion of enabled +// Routing target checks and their referenced upstreams. TARGET scheduling +// materializes one bounded due reference per live proxy and target profile. +const MaximumTargetProfilesPerUpstream = 64 + const ( MaximumPoolSize = 1_000_000 MaximumExactCounter = int64(1<<53 - 1) @@ -182,6 +187,13 @@ type Routing struct { Upstreams []string `yaml:"upstreams"` Strategy Strategy `yaml:"strategy"` OnUnavailable OnUnavailable `yaml:"onUnavailable"` + Check RoutingCheck `yaml:"check"` +} + +// RoutingCheck defines target-specific health probes. Unlike upstream Check, +// these facts are isolated by Routing and never mutate global proxy health. +type RoutingCheck struct { + Targets []string `yaml:"targets"` } type RoutingMatch struct { diff --git a/internal/config/redact.go b/internal/config/redact.go index 52c4478..450a473 100644 --- a/internal/config/redact.go +++ b/internal/config/redact.go @@ -107,6 +107,7 @@ func cloneRouting(source Routing) Routing { cloned.Match.Headers = cloneStringMap(source.Match.Headers) cloned.Upstreams = cloneStrings(source.Upstreams) cloned.Strategy.Weights = cloneIntMap(source.Strategy.Weights) + cloned.Check.Targets = cloneStrings(source.Check.Targets) return cloned } diff --git a/internal/config/routing_check_test.go b/internal/config/routing_check_test.go new file mode 100644 index 0000000..00f46bc --- /dev/null +++ b/internal/config/routing_check_test.go @@ -0,0 +1,80 @@ +package config + +import ( + "fmt" + "strings" + "testing" +) + +func TestValidateRoutingCheckTargets(t *testing.T) { + source := strings.Replace(validConfig, " strategy:", " check:\n targets: [https://checkout.example/health]\n strategy:", 1) + loaded, err := Load(strings.NewReader(source)) + if err != nil || len(loaded.Routing[0].Check.Targets) != 1 || loaded.Routing[0].Check.Targets[0] != "https://checkout.example/health" { + t.Fatalf("Load(routing check targets) = (%+v, %v)", loaded, err) + } + + cfg := mustLoadValidConfig(t) + route := cfg.Routing[0] + route.Check.Targets = []string{"https://checkout.example/health"} + cfg.Routing[0] = route + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(routing check targets) = %v", err) + } + + redacted := cfg.Redacted() + redacted.Routing[0].Check.Targets[0] = "https://changed.example/health" + if cfg.Routing[0].Check.Targets[0] == redacted.Routing[0].Check.Targets[0] { + t.Fatal("Redacted() routing check targets alias source configuration") + } +} + +func TestValidateRejectsInvalidOrUnboundedRoutingCheckTargets(t *testing.T) { + for _, targets := range [][]string{ + {"https://checkout.example/health", "https://checkout.example/health"}, + {"ftp://checkout.example/health"}, + } { + cfg := mustLoadValidConfig(t) + route := cfg.Routing[0] + route.Check.Targets = targets + cfg.Routing[0] = route + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "check.targets") { + t.Fatalf("Validate(targets=%v) error = %v, want check.targets", targets, err) + } + } + + cfg := mustLoadValidConfig(t) + route := cfg.Routing[0] + route.Name = "checkout route" + route.Check.Targets = []string{"https://checkout.example/health"} + cfg.Routing[0] = route + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "target profile identifier") { + t.Fatalf("Validate(invalid target routing name) error = %v", err) + } + + cfg = mustLoadValidConfig(t) + route = cfg.Routing[0] + route.Check.Targets = make([]string, MaximumCheckURLs+1) + for index := range route.Check.Targets { + route.Check.Targets[index] = fmt.Sprintf("https://target-%d.example/health", index) + } + cfg.Routing[0] = route + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "at most") { + t.Fatalf("Validate(too many route targets) error = %v", err) + } + + cfg = mustLoadValidConfig(t) + route = cfg.Routing[0] + route.Check.Targets = make([]string, MaximumCheckURLs) + for index := range route.Check.Targets { + route.Check.Targets[index] = fmt.Sprintf("https://target-%d.example/health", index) + } + cfg.Routing[0] = route + for index := 0; index < 4; index++ { + copyRoute := route + copyRoute.Name = fmt.Sprintf("route-%d", index) + cfg.Routing = append(cfg.Routing, copyRoute) + } + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "routing target profiles") { + t.Fatalf("Validate(unbounded target profiles) error = %v", err) + } +} diff --git a/internal/config/validate.go b/internal/config/validate.go index e1bccff..9fec7af 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -74,10 +74,24 @@ func Validate(cfg *Config) error { return fmt.Errorf("validate configuration: at least one enabled upstream is required") } seen := make(map[string]struct{}, len(cfg.Routing)) + targetProfilesByUpstream := make(map[string]int, len(cfg.Upstreams)) for index, route := range cfg.Routing { if err := validateRouting(index, route, cfg.Upstreams, seen); err != nil { return err } + if !route.Enabled || len(route.Check.Targets) == 0 { + continue + } + for _, upstreamID := range route.Upstreams { + upstream, exists := cfg.Upstreams[upstreamID] + if !exists || !upstream.Enabled { + continue + } + targetProfilesByUpstream[upstreamID] += len(route.Check.Targets) + if targetProfilesByUpstream[upstreamID] > MaximumTargetProfilesPerUpstream { + return fmt.Errorf("validate upstream %q routing target profiles: supports at most %d", upstreamID, MaximumTargetProfilesPerUpstream) + } + } } if cfg.Distribution.Enabled { clientIdentificationMode := cfg.Distribution.ClientIdentification.Mode @@ -272,6 +286,9 @@ func validateRouting(index int, route Routing, upstreams map[string]Upstream, se } seen[route.Name] = struct{}{} scope := fmt.Sprintf("routing %q", route.Name) + if len(route.Check.Targets) > 0 && !validTargetRoutingName(route.Name) { + return fmt.Errorf("validate %s check.targets: routing name is not a valid target profile identifier", scope) + } if err := validateEnum(scope+" purpose", route.Purpose, "gateway", "extract"); err != nil { return err } @@ -302,6 +319,9 @@ func validateRouting(index int, route Routing, upstreams map[string]Upstream, se if route.OnUnavailable.Action == "wait" && route.OnUnavailable.WaitTimeout <= 0 { return fmt.Errorf("validate %s onUnavailable.waitTimeout: must be greater than zero for wait", scope) } + if err := validateCheckURLs(scope+" check.targets", route.Check.Targets); err != nil { + return err + } return nil } @@ -588,28 +608,44 @@ func validateCheck(scope string, check Check) error { if err := requirePositive(scope+".maxConsecutiveFailures", check.MaxConsecutiveFailures); err != nil { return err } - if len(check.URLs) > MaximumCheckURLs { - return fmt.Errorf("validate %s.urls: supports at most %d URLs", scope, MaximumCheckURLs) + return validateCheckURLs(scope+".urls", check.URLs) +} + +func validateCheckURLs(scope string, urls []string) error { + if len(urls) > MaximumCheckURLs { + return fmt.Errorf("validate %s: supports at most %d URLs", scope, MaximumCheckURLs) } - seenURLs := make(map[string]struct{}, len(check.URLs)) - for index, rawURL := range check.URLs { + seenURLs := make(map[string]struct{}, len(urls)) + for index, rawURL := range urls { if rawURL == "" || strings.TrimSpace(rawURL) != rawURL { - return fmt.Errorf("validate %s.urls[%d]: must be an absolute http or https URL", scope, index) + return fmt.Errorf("validate %s[%d]: must be an absolute http or https URL", scope, index) } parsed, err := url.Parse(rawURL) if err != nil || parsed.Scheme == "" || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" || (parsed.Scheme != "http" && parsed.Scheme != "https") { - return fmt.Errorf("validate %s.urls[%d]: must be an absolute http or https URL", scope, index) + return fmt.Errorf("validate %s[%d]: must be an absolute http or https URL", scope, index) } canonical := parsed.String() if _, exists := seenURLs[canonical]; exists { - return fmt.Errorf("validate %s.urls[%d]: duplicate URL", scope, index) + return fmt.Errorf("validate %s[%d]: duplicate URL", scope, index) } seenURLs[canonical] = struct{}{} } return nil } +func validTargetRoutingName(value string) bool { + if value == "" || len(value) > 256 || strings.TrimSpace(value) != value { + return false + } + for _, character := range value { + if character <= ' ' || character == '\x7f' { + return false + } + } + return true +} + func fetchConfigured(fetch Fetch) bool { return fetch.EstimatedIPsPerCall != 0 || fetch.RequestInterval != 0 || fetch.Timeout != 0 || fetch.MaxAttempts != 0 || fetch.MaxInFlight != 0 || fetch.MaxTotal != 0 || fetch.MaxResponseBytes != 0 || diff --git a/internal/controller/health/scheduler_runner.go b/internal/controller/health/scheduler_runner.go index 07556cb..c2622f4 100644 --- a/internal/controller/health/scheduler_runner.go +++ b/internal/controller/health/scheduler_runner.go @@ -9,6 +9,7 @@ import ( "time" "proxy-pool/internal/config" + healthDomain "proxy-pool/internal/domain/health" ) var ( @@ -39,6 +40,13 @@ type EgressUpstreamTaskSource interface { DueEgressCandidatesForUpstream(context.Context, string, string, time.Time, int) ([]Candidate, error) } +// TargetUpstreamTaskSource supplies bounded per-upstream candidates for one +// Routing target profile. TARGET facts remain separate from global health. +type TargetUpstreamTaskSource interface { + UpstreamTaskSource + DueTargetCandidatesForUpstream(context.Context, string, string, string, time.Time, int) ([]Candidate, error) +} + type upstreamDueSource struct { source UpstreamTaskSource upstreamID string @@ -80,6 +88,38 @@ func (source egressUpstreamDueSource) DueCandidates(ctx context.Context, now tim return source.source.DueEgressCandidatesForUpstream(ctx, source.upstreamID, source.targetURL, now, limit) } +type targetUpstreamDueSource struct { + source TargetUpstreamTaskSource + upstreamID string + routingName string + targetURL string +} + +func newTargetUpstreamDueSource( + source TargetUpstreamTaskSource, + upstreamID string, + routingName string, + targetURL string, +) (DueSource, error) { + profile, err := healthDomain.NormalizeTargetProfile(healthDomain.TargetProfile{ + RoutingName: routingName, TargetURL: targetURL, + }) + if nilInterface(source) || upstreamID == "" || err != nil || profile.RoutingName != routingName || profile.TargetURL != targetURL { + return nil, ErrInvalidSchedulerRunner + } + return targetUpstreamDueSource{ + source: source, upstreamID: upstreamID, routingName: routingName, targetURL: targetURL, + }, nil +} + +func (source targetUpstreamDueSource) InFlight(ctx context.Context, now time.Time) (int, error) { + return source.source.InFlightForUpstream(ctx, source.upstreamID, now) +} + +func (source targetUpstreamDueSource) DueCandidates(ctx context.Context, now time.Time, limit int) ([]Candidate, error) { + return source.source.DueTargetCandidatesForUpstream(ctx, source.upstreamID, source.routingName, source.targetURL, now, limit) +} + // TaskSink atomically offers an already bounded batch to the shared leased // task store. It must leave unaccepted candidates eligible for a later tick. type TaskSink interface { @@ -224,14 +264,18 @@ func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (Tick return TickResult{}, err } check := config.EffectiveCheck(configuration.Defaults.Check, configuration.Upstreams[name].Check) - groups, err := supervisor.upstreamDueGroups(basicSource, name, check.URLs) + groups, err := supervisor.upstreamDueGroups(basicSource, name, check.URLs, configuredTargetProfiles(configuration, name)) if err != nil { return TickResult{}, err } - groupOptions := schedulerOptionsForPlanner(supervisor.options, planner) - groupOptions.BatchSize = schedulerGroupBatchSize(groupOptions.BatchSize, len(groups)) + remaining := schedulerOptionsForPlanner(supervisor.options, planner).BatchSize start := int(supervisor.cursor.Add(1)-1) % len(groups) for offset := range groups { + if remaining <= 0 { + break + } + groupOptions := schedulerOptionsForPlanner(supervisor.options, planner) + groupOptions.BatchSize = schedulerGroupBatchSize(remaining, len(groups)-offset) source := groups[(start+offset)%len(groups)] tick, err := tickWithPlanner(ctx, planner, source, supervisor.sink, groupOptions) if err != nil { @@ -239,27 +283,76 @@ func (supervisor *ConfiguredSchedulerSupervisor) Tick(ctx context.Context) (Tick } result.Planned += tick.Planned result.Offered += tick.Offered + remaining -= tick.Offered } } return result, nil } -func (supervisor *ConfiguredSchedulerSupervisor) upstreamDueGroups(basic DueSource, upstreamID string, urls []string) ([]DueSource, error) { +func (supervisor *ConfiguredSchedulerSupervisor) upstreamDueGroups( + basic DueSource, + upstreamID string, + egressURLs []string, + targets []configuredTargetProfile, +) ([]DueSource, error) { groups := []DueSource{basic} - egress, supported := supervisor.source.(EgressUpstreamTaskSource) - if !supported || len(urls) == 0 { - return groups, nil - } - for _, targetURL := range urls { - source, err := newEgressUpstreamDueSource(egress, upstreamID, targetURL) - if err != nil { - return nil, err + if egress, supported := supervisor.source.(EgressUpstreamTaskSource); supported { + for _, targetURL := range egressURLs { + source, err := newEgressUpstreamDueSource(egress, upstreamID, targetURL) + if err != nil { + return nil, err + } + groups = append(groups, source) + } + } + if target, supported := supervisor.source.(TargetUpstreamTaskSource); supported { + for _, profile := range targets { + source, err := newTargetUpstreamDueSource(target, upstreamID, profile.RoutingName, profile.TargetURL) + if err != nil { + return nil, err + } + groups = append(groups, source) } - groups = append(groups, source) } return groups, nil } +type configuredTargetProfile struct { + RoutingName string + TargetURL string +} + +func configuredTargetProfiles(configuration *config.Config, upstreamID string) []configuredTargetProfile { + if configuration == nil || upstreamID == "" { + return nil + } + profiles := make([]configuredTargetProfile, 0) + for _, route := range configuration.Routing { + if !route.Enabled || !containsString(route.Upstreams, upstreamID) { + continue + } + for _, targetURL := range route.Check.Targets { + profiles = append(profiles, configuredTargetProfile{RoutingName: route.Name, TargetURL: targetURL}) + } + } + sort.Slice(profiles, func(left, right int) bool { + if profiles[left].RoutingName != profiles[right].RoutingName { + return profiles[left].RoutingName < profiles[right].RoutingName + } + return profiles[left].TargetURL < profiles[right].TargetURL + }) + return profiles +} + +func containsString(values []string, target string) bool { + for _, value := range values { + if value == target { + return true + } + } + return false +} + func schedulerGroupBatchSize(batchSize, groups int) int { if batchSize <= 0 || groups <= 0 { return 0 diff --git a/internal/controller/health/scheduler_runner_test.go b/internal/controller/health/scheduler_runner_test.go index 5a289f0..0e56296 100644 --- a/internal/controller/health/scheduler_runner_test.go +++ b/internal/controller/health/scheduler_runner_test.go @@ -194,6 +194,82 @@ func TestConfiguredSchedulerSupervisorSchedulesBoundedEgressGroups(t *testing.T) } } +func TestConfiguredSchedulerSupervisorSchedulesRoutingTargetProfiles(t *testing.T) { + now := time.Date(2026, 8, 2, 11, 30, 0, 0, time.UTC) + const routingName = "checkout" + const targetURL = "https://checkout.example/health" + configuration := &config.Config{ + Defaults: config.Defaults{Check: config.Check{ + Interval: config.Duration(time.Minute), MaxInFlight: 4, Timeout: config.Duration(time.Second), MaxAttempts: 1, + }}, + Routing: []config.Routing{{ + Name: routingName, Enabled: true, Upstreams: []string{"provider-a"}, + Check: config.RoutingCheck{Targets: []string{targetURL}}, + }}, + Upstreams: map[string]config.Upstream{"provider-a": {Enabled: true}}, + } + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": {candidates: []Candidate{{ + ProxyID: "basic", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now, + }}, targetCandidates: map[string][]Candidate{ + targetCandidateKey(routingName, targetURL): {{ + ProxyID: "target", State: proxyDomain.StateFetched, Level: healthDomain.LevelTarget, + RoutingName: routingName, TargetURL: targetURL, DueAt: now, + }}, + }}, + }} + sink := &taskSinkStub{} + supervisor, err := NewConfiguredSchedulerSupervisor(&configurationSourceStub{configuration: configuration}, source, sink, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + result, err := supervisor.Tick(context.Background()) + if err != nil || result != (TickResult{Planned: 2, Offered: 2}) || len(sink.tasks) != 1 || + sink.tasks[0].Candidate.Level != healthDomain.LevelTarget || sink.tasks[0].Candidate.RoutingName != routingName || + source.sources["provider-a"].targetCalls != 1 { + t.Fatalf("Tick() = (%+v, %v); source=%+v sink=%+v", result, err, source, sink.tasks) + } +} + +func TestConfiguredSchedulerSupervisorNeverExceedsTotalBatchAcrossGroups(t *testing.T) { + now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) + configuration := &config.Config{ + Defaults: config.Defaults{Check: config.Check{ + Interval: config.Duration(time.Minute), MaxInFlight: 10, Timeout: config.Duration(time.Second), MaxAttempts: 1, + URLs: []string{"https://egress.example/identity"}, + }}, + Routing: []config.Routing{{ + Name: "checkout", Enabled: true, Upstreams: []string{"provider-a"}, + Check: config.RoutingCheck{Targets: []string{"https://checkout.example/health"}}, + }}, + Upstreams: map[string]config.Upstream{"provider-a": {Enabled: true}}, + } + source := &upstreamTaskSourceStub{sources: map[string]*dueSourceStub{ + "provider-a": { + candidates: []Candidate{ + {ProxyID: "basic-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}, + {ProxyID: "basic-b", State: proxyDomain.StateFetched, Level: healthDomain.LevelBasic, DueAt: now}, + }, + egressCandidates: map[string][]Candidate{"https://egress.example/identity": { + {ProxyID: "egress-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelEgress, TargetURL: "https://egress.example/identity", DueAt: now}, + }}, + targetCandidates: map[string][]Candidate{targetCandidateKey("checkout", "https://checkout.example/health"): { + {ProxyID: "target-a", State: proxyDomain.StateFetched, Level: healthDomain.LevelTarget, RoutingName: "checkout", TargetURL: "https://checkout.example/health", DueAt: now}, + }}, + }, + }} + supervisor, err := NewConfiguredSchedulerSupervisor(&configurationSourceStub{configuration: configuration}, source, &taskSinkStub{}, + SchedulerRunnerOptions{PollInterval: time.Second, BatchSize: 4, Now: func() time.Time { return now }}) + if err != nil { + t.Fatalf("NewConfiguredSchedulerSupervisor() = %v", err) + } + result, err := supervisor.Tick(context.Background()) + if err != nil || result.Planned != 4 || result.Offered != 4 { + t.Fatalf("Tick() = (%+v, %v), want exactly four planned/offered tasks", result, err) + } +} + type dueSourceStub struct { inFlight int candidates []Candidate @@ -202,6 +278,8 @@ type dueSourceStub struct { dueCalls int egressCandidates map[string][]Candidate egressCalls int + targetCandidates map[string][]Candidate + targetCalls int } func (source *dueSourceStub) InFlight(context.Context, time.Time) (int, error) { @@ -221,6 +299,18 @@ func (source *dueSourceStub) DueEgressCandidates(_ context.Context, targetURL st return source.egressCandidates[targetURL], nil } +func (source *dueSourceStub) DueTargetCandidates( + _ context.Context, + routingName string, + targetURL string, + _ time.Time, + limit int, +) ([]Candidate, error) { + source.targetCalls++ + source.limit = limit + return source.targetCandidates[targetCandidateKey(routingName, targetURL)], nil +} + type upstreamTaskSourceStub struct { sources map[string]*dueSourceStub } @@ -249,6 +339,25 @@ func (source *upstreamTaskSourceStub) DueEgressCandidatesForUpstream(ctx context return item.DueEgressCandidates(ctx, targetURL, now, limit) } +func (source *upstreamTaskSourceStub) DueTargetCandidatesForUpstream( + ctx context.Context, + upstreamID string, + routingName string, + targetURL string, + now time.Time, + limit int, +) ([]Candidate, error) { + item, exists := source.sources[upstreamID] + if !exists { + return nil, errors.New("missing upstream source") + } + return item.DueTargetCandidates(ctx, routingName, targetURL, now, limit) +} + +func targetCandidateKey(routingName, targetURL string) string { + return routingName + "\x00" + targetURL +} + type taskSinkStub struct { tasks []PlannedTask offer int