diff --git a/README.md b/README.md index 597948c..c87b6f7 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ flowchart LR Controller 的 Redis 共享 BASIC/EGRESS 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 Checker 探测和 Observation 状态归并。 -- **部分完成**:EGRESS 的出口身份响应解析、TARGET 的任务编排和配置建模,Docker Compose/Kubernetes +- **部分完成**:TARGET 的任务编排和配置建模,Docker Compose/Kubernetes 运行时 mTLS Overlay。 - **待完成**:CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。 @@ -172,7 +172,8 @@ Controller 在启用控制面时装配 Redis 共享任务 broker,并按启用 HTTP/HTTPS/SOCKS5 BASIC 检查,并按每个 `check.urls` 创建 EGRESS 任务。调度监督器每轮读取 已发布配置,因此 reload 后的上游启停、检查间隔、抖动、超时、重试次数、`maxInFlight` 和 EGRESS URL 都会在下一轮生效;BASIC 与 EGRESS 以有界轮转组共享上游并发上限。新启用的上游 -无需重启 Controller。TARGET 尚未进入生产调度,EGRESS 的出口身份响应解析也尚未配置化。 +无需重启 Controller。EGRESS 对成功响应提取纯文本 IP 或常见 JSON IP 字段,并将其作为全局 +健康事实回传;TARGET 尚未进入生产调度。 初版 HTTP 容量工具可按固定请求数或固定时长运行,并将 HTTPS 目标经 Gateway 的请求 交给标准 HTTP Transport 建立 CONNECT: @@ -229,8 +230,7 @@ go run ./cmd/proxy-loadgen ` - **P0 - Worker 控制面闭环**:Worker session、Snapshot ledger、ACK、运行态接收、 ownership 索引,以及 Gateway 快照客户端。 - **P0 - Checker 健康链**:BASIC/EGRESS 的共享调度、实际探测、Observation reducer 和 - `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成;继续补齐 EGRESS 身份解析、 - TARGET。 + `FETCHED -> AVAILABLE / SUSPECT / UNHEALTHY` 状态链已完成;继续补齐 TARGET。 - **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 4ddbcfc..1002830 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -185,7 +185,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 的出口身份响应解析与 EGRESS/TARGET 生产调度仍待后续实现。 +非成功状态作为事实。EGRESS 对成功响应解析纯文本或常见 JSON IP 字段;EGRESS 生产调度 +已完成,TARGET 生产调度仍待后续实现。 ## 7. 兼容与演进 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index d8e6e80..eec0579 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -465,7 +465,8 @@ proxyAuth: - 距离过期不足 `allocationSafetyMargin` 时停止新分配。 - `check.jitter` 为调度抖动百分比,避免所有 Proxy 同时探测。 - `check.urls` 最多 16 个规范化 HTTP/HTTPS URL,作为 EGRESS 检查的有界目标集; - BASIC 检查不依赖该字段。 + BASIC 检查不依赖该字段。EGRESS 成功响应必须包含合法 IP:支持纯文本 IP,或 JSON 的 + `ip`、`ip_address`、`address`、`origin`、`query` 字段。 - 第一次有意义失败进入 SUSPECT;达到 `maxConsecutiveFailures` 后才进入 UNHEALTHY。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 5a32ba0..c814bf8 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -262,7 +262,7 @@ Gateway 请求热路径仍只使用本地准入,不增加 Redis/PostgreSQL 调 WorkerControlPlane gRPC 接收端、session 签发/心跳、Snapshot ACK 账本、基础 Snapshot 流、Gateway 会话客户端与快照凭据分发已完成;权威 Proxy/Routing 发布、 Outcome 上报已完成为有界队列、批次序列/摘要栅栏和确认重试;健康 BASIC/EGRESS 执行链已 -完成,EGRESS 身份解析、TARGET 编排和 REMOVE 生命周期仍待完成,因此 Task 10 尚未全部完成。 +完成,TARGET 编排和 REMOVE 生命周期仍待完成,因此 Task 10 尚未全部完成。 ## Task 11: Checker and Health Reducer @@ -286,8 +286,8 @@ 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 的出口身份响应解析与部署运行态 -仍未实现, +TARGET 探测器具备任务执行能力,但尚无生产任务调度;EGRESS 的出口身份响应解析已支持固定 +上限的纯文本和常见 JSON IP 字段,部署运行态仍未实现, 因此本任务保持未完成。 补充进度(2026-08-02):BASIC 调度已改为配置驱动监督器。它每轮读取已发布快照并复用 diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index 7aa2cee..d8998d9 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -25,7 +25,7 @@ Distribution/Admin/Metrics 独立监听和有界停机装配。Provider 自动补池、分布式 配额、动态重载和 Admin 低基数统计已装配;Controller 已装配 Redis BASIC/EGRESS 任务 broker, `proxy-checker` 可执行 HTTP/HTTPS/SOCKS5 BASIC/EGRESS 探测。`proxy-loadgen` 已提供有界 HTTP -请求场景;EGRESS 身份解析、TARGET 调度、CONNECT 长连接/Extract 压测与完整 mTLS 环境 Overlay 仍属于 +请求场景;TARGET 调度、CONNECT 长连接/Extract 压测与完整 mTLS 环境 Overlay 仍属于 `implementation-plan.md` 后续任务。 因此 Compose/Kubernetes 资产当前仍用于评审网络、资源、探针和依赖关系,不能 视为完整可运行拓扑。 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index cbba61f..45d3b71 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -63,7 +63,7 @@ Outcome 已实现为 Gateway 本地有界队列、微批确认重试和 Controll 归并边界,Checker Observation RPC 已在同一控制面监听以独立 SPIFFE 身份接入;有界任务领取、 租约归属与任务期凭据传输已由通用契约和 gRPC 往返测试覆盖。Redis 共享 due-index、 任务/租约持久化、按上游的 in-flight 限制以及独立 Checker 的 HTTP/HTTPS/SOCKS5 BASIC/EGRESS 执行 -进程已经闭环;EGRESS 已具备多维任务索引与生产调度,出口身份响应解析和 TARGET 的多维索引仍未实现。Snapshot 签发在 Redis 中原子匹配当前 +进程已经闭环;EGRESS 已具备多维任务索引、出口身份响应解析与生产调度,TARGET 的多维索引仍未实现。Snapshot 签发在 Redis 中原子匹配当前 `session_id`,重注册会清除旧引用,迟到旧 Stream 不会覆盖新 session。Controller 在最近成功下发的 Snapshot `valid_until` 到达时关闭流;Gateway 的公用 `SessionSupervisor` 已实现可恢复错误的有界退避重连。Gateway 会校验并执行 Snapshot diff --git a/internal/checker/controlplane/runner.go b/internal/checker/controlplane/runner.go index 6008c93..cc7d5c6 100644 --- a/internal/checker/controlplane/runner.go +++ b/internal/checker/controlplane/runner.go @@ -213,7 +213,7 @@ func (runner *Runner) executeTask(ctx context.Context, task *controlplanev1.Chec } observation := &controlplanev1.HealthObservation{ TaskId: task.GetTaskId(), LeaseToken: task.GetLeaseToken(), ProxyId: task.GetProxyId(), Level: task.GetLevel(), Success: result.Success, - FailureClass: result.FailureClass, Latency: durationpb.New(result.Latency), + FailureClass: result.FailureClass, ObservedEgressIp: result.ObservedEgressIP, Latency: durationpb.New(result.Latency), ObservedAt: timestamppb.New(runner.options.Now().UTC()), } if task.GetLevel() == controlplanev1.CheckLevel_CHECK_LEVEL_TARGET { diff --git a/internal/checker/controlplane/runner_test.go b/internal/checker/controlplane/runner_test.go index 2b91ba9..df8d55b 100644 --- a/internal/checker/controlplane/runner_test.go +++ b/internal/checker/controlplane/runner_test.go @@ -62,6 +62,9 @@ func TestRunnerOmitsEgressProbeURLFromGlobalObservation(t *testing.T) { if observation.GetLevel() != controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS || observation.GetTargetUrl() != "" || observation.GetRoutingName() != "" { t.Fatalf("Observation = %+v, want global EGRESS fact without target profile", observation) } + if observation.GetObservedEgressIp() != "198.51.100.42" { + t.Fatalf("Observation egress IP = %q", observation.GetObservedEgressIp()) + } } func checkerTask(id string, now time.Time) *controlplanev1.CheckTask { @@ -117,5 +120,9 @@ func (stub *executorStub) Execute(_ context.Context, task *controlplanev1.CheckT if task.GetTaskId() == "task-a" && stub.calls[task.GetTaskId()] == 1 { return probe.Result{FailureClass: probe.FailureProxyRequest, Latency: time.Millisecond} } - return probe.Result{Success: true, Latency: 2 * time.Millisecond} + result := probe.Result{Success: true, Latency: 2 * time.Millisecond} + if task.GetLevel() == controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS { + result.ObservedEgressIP = "198.51.100.42" + } + return result } diff --git a/internal/checker/probe/probe.go b/internal/checker/probe/probe.go index 1afefd7..ed7611b 100644 --- a/internal/checker/probe/probe.go +++ b/internal/checker/probe/probe.go @@ -4,9 +4,12 @@ package probe import ( "context" + "encoding/json" "errors" + "io" "net" "net/http" + "net/netip" "net/url" "strconv" "strings" @@ -21,16 +24,19 @@ const ( FailureProxyRequest = "PROXY_REQUEST" FailureProxyAuth = "PROXY_AUTH" FailureTargetHTTPStatus = "TARGET_HTTP_STATUS" + FailureEgressIdentity = "EGRESS_IDENTITY" FailureUnsupportedProxy = "UNSUPPORTED_PROXY_PROTOCOL" basicHandshakeProbeURL = "http://example.invalid/" + maximumEgressBodyBytes = 16 << 10 ) var errUnsupportedProtocol = errors.New("unsupported proxy protocol") type Result struct { - Success bool - FailureClass string - Latency time.Duration + Success bool + FailureClass string + ObservedEgressIP string + Latency time.Duration } // Executor constructs a short-lived HTTP transport for each task. That keeps @@ -52,9 +58,9 @@ func (executor *Executor) Execute(ctx context.Context, task *controlplanev1.Chec deadline, target, proxyURL, err := prepare(task, started.UTC()) if err != nil { if errors.Is(err, errUnsupportedProtocol) { - return Result{FailureClass: FailureUnsupportedProxy, Latency: time.Since(started)} + return Result{FailureClass: FailureUnsupportedProxy, Latency: probeLatency(started)} } - return Result{FailureClass: FailureInvalidTask, Latency: time.Since(started)} + return Result{FailureClass: FailureInvalidTask, Latency: probeLatency(started)} } probeContext, cancel := context.WithDeadline(ctx, deadline) defer cancel() @@ -76,10 +82,10 @@ func (executor *Executor) Execute(ctx context.Context, task *controlplanev1.Chec } request, err := http.NewRequestWithContext(probeContext, http.MethodGet, target, nil) if err != nil { - return Result{FailureClass: FailureInvalidTask, Latency: time.Since(started)} + return Result{FailureClass: FailureInvalidTask, Latency: probeLatency(started)} } response, err := client.Do(request) - latency := time.Since(started) + latency := probeLatency(started) if err != nil { if errors.Is(probeContext.Err(), context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded) { return Result{FailureClass: FailureDeadlineExceeded, Latency: latency} @@ -89,7 +95,7 @@ func (executor *Executor) Execute(ctx context.Context, task *controlplanev1.Chec } return Result{FailureClass: FailureProxyRequest, Latency: latency} } - _ = response.Body.Close() + defer response.Body.Close() if response.StatusCode == http.StatusProxyAuthRequired { return Result{FailureClass: FailureProxyAuth, Latency: latency} } @@ -99,9 +105,56 @@ func (executor *Executor) Execute(ctx context.Context, task *controlplanev1.Chec if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusBadRequest { return Result{FailureClass: FailureTargetHTTPStatus, Latency: latency} } + if task.GetLevel() == controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS { + body, err := io.ReadAll(io.LimitReader(response.Body, maximumEgressBodyBytes+1)) + if err != nil || len(body) > maximumEgressBodyBytes { + return Result{FailureClass: FailureEgressIdentity, Latency: latency} + } + observedIP, ok := parseEgressIP(body) + if !ok { + return Result{FailureClass: FailureEgressIdentity, Latency: latency} + } + return Result{Success: true, ObservedEgressIP: observedIP, Latency: latency} + } return Result{Success: true, Latency: latency} } +func parseEgressIP(body []byte) (string, bool) { + if observed, ok := normalizeEgressIP(string(body)); ok { + return observed, true + } + var payload map[string]json.RawMessage + if err := json.Unmarshal(body, &payload); err != nil { + return "", false + } + for _, key := range []string{"ip", "ip_address", "address", "origin", "query"} { + raw, exists := payload[key] + if !exists { + continue + } + var value string + if err := json.Unmarshal(raw, &value); err != nil { + continue + } + if observed, ok := normalizeEgressIP(value); ok { + return observed, true + } + } + return "", false +} + +func normalizeEgressIP(value string) (string, bool) { + for _, item := range strings.FieldsFunc(strings.TrimSpace(value), func(character rune) bool { + return character == ',' || character == ';' || character == ' ' || character == '\t' || character == '\r' || character == '\n' + }) { + address, err := netip.ParseAddr(item) + if err == nil { + return address.Unmap().String(), true + } + } + return "", false +} + func prepare(task *controlplanev1.CheckTask, now time.Time) (time.Time, string, *url.URL, error) { if task == nil || task.GetTaskId() == "" || task.GetProxyId() == "" || task.GetHost() == "" || task.GetPort() == 0 || task.GetTimeout() == nil || task.GetTimeout().CheckValid() != nil || task.GetTimeout().AsDuration() <= 0 || @@ -171,3 +224,10 @@ func minDuration(left, right time.Duration) time.Duration { } return right } + +func probeLatency(started time.Time) time.Duration { + if latency := time.Since(started); latency > 0 { + return latency + } + return time.Nanosecond +} diff --git a/internal/checker/probe/probe_test.go b/internal/checker/probe/probe_test.go index bf00334..23e0beb 100644 --- a/internal/checker/probe/probe_test.go +++ b/internal/checker/probe/probe_test.go @@ -77,6 +77,56 @@ func TestExecutorRejectsEgressTaskWithRoutingProfile(t *testing.T) { } } +func TestExecutorEgressReportsObservedIP(t *testing.T) { + proxy := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.URL.String() != "http://egress.example/identity" { + t.Errorf("proxy request URL = %q", request.URL) + } + response.Header().Set("Content-Type", "application/json") + _, _ = response.Write([]byte(`{"ip":"2001:db8::42"}`)) + })) + defer proxy.Close() + task := validTask(t, proxy.URL, controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS) + task.TargetUrl = "http://egress.example/identity" + result := NewExecutor().Execute(context.Background(), task) + if !result.Success || result.FailureClass != "" || result.ObservedEgressIP != "2001:db8::42" || result.Latency <= 0 { + t.Fatalf("Execute(EGRESS) = %+v", result) + } +} + +func TestExecutorEgressRejectsMissingIdentity(t *testing.T) { + proxy := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, _ *http.Request) { + _, _ = response.Write([]byte(`{"status":"ok"}`)) + })) + defer proxy.Close() + task := validTask(t, proxy.URL, controlplanev1.CheckLevel_CHECK_LEVEL_EGRESS) + task.TargetUrl = "http://egress.example/identity" + result := NewExecutor().Execute(context.Background(), task) + if result.Success || result.FailureClass != FailureEgressIdentity || result.ObservedEgressIP != "" || result.Latency <= 0 { + t.Fatalf("Execute(EGRESS missing identity) = %+v", result) + } +} + +func TestParseEgressIP(t *testing.T) { + for _, test := range []struct { + name string + body string + want string + }{ + {name: "plain IPv4", body: "198.51.100.7\n", want: "198.51.100.7"}, + {name: "JSON query", body: `{"query":"2001:db8::7"}`, want: "2001:db8::7"}, + {name: "JSON origin list", body: `{"origin":"198.51.100.8, 2001:db8::8"}`, want: "198.51.100.8"}, + {name: "invalid", body: `{"ip":"unknown"}`}, + } { + t.Run(test.name, func(t *testing.T) { + got, ok := parseEgressIP([]byte(test.body)) + if got != test.want || ok != (test.want != "") { + t.Fatalf("parseEgressIP(%q) = (%q, %t), want (%q, %t)", test.body, got, ok, test.want, test.want != "") + } + }) + } +} + func TestExecutorReportsUnsupportedProtocolAsFact(t *testing.T) { now := time.Now().UTC() result := NewExecutor().Execute(context.Background(), &controlplanev1.CheckTask{