From d04b7413eb394d2815f24e32474bbef1030c7943 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 09:33:21 +0800 Subject: [PATCH] feat: add load generator connect scenario --- README.md | 20 +- cmd/proxy-loadgen/main.go | 9 +- cmd/proxy-loadgen/main_test.go | 16 ++ docs/development/implementation-plan.md | 11 +- docs/requirements/completion-audit.md | 4 +- docs/requirements/traceability.md | 4 +- internal/loadgen/http.go | 294 +++++++++++++++++++----- internal/loadgen/http_test.go | 78 +++++++ 8 files changed, 363 insertions(+), 73 deletions(-) diff --git a/README.md b/README.md index 85f10a1..dba91e9 100644 --- a/README.md +++ b/README.md @@ -177,8 +177,8 @@ BASIC、EGRESS 与 TARGET 以有界轮转组共享上游并发上限。新启用 EGRESS 对成功响应提取纯文本 IP 或常见 JSON IP 字段，并将其作为全局健康事实回传；TARGET 事实 仅归并到对应的 `(routing_name, target_url)` Profile，不改变 Proxy 全局健康。 -初版 HTTP 容量工具可按固定请求数或固定时长运行，并将 HTTPS 目标经 Gateway 的请求 -交给标准 HTTP Transport 建立 CONNECT： +`proxy-loadgen` 的 HTTP 场景可按固定请求数或固定时长运行，并将 HTTPS 目标经 +Gateway 的请求交给标准 HTTP Transport 建立 CONNECT： ```powershell go run ./cmd/proxy-loadgen ` @@ -187,6 +187,16 @@ go run ./cmd/proxy-loadgen ` -requests 10000 -concurrency 128 -timeout 10s ``` +CONNECT 长连接场景直接向 HTTP Gateway 发起隧道握手，并在成功后保持每条隧道指定 +时长；`-timeout` 只限制 TCP 连接与 CONNECT 响应，`-hold` 控制建连后的保持时间： + +```powershell +go run ./cmd/proxy-loadgen ` + -scenario connect -target https://TARGET_URL/ ` + -proxy http://GATEWAY_HOST:8080 ` + -requests 1000 -concurrency 128 -timeout 5s -hold 30s +``` + 使用 `-duration 30s -rate 5000` 可运行限速场景；省略 `-rate` 时固定数量 worker 会饱和发送。`-method`、重复的 `-header` 与 `-body` 可组合用于 Distribution 的提取接口： @@ -198,9 +208,9 @@ go run ./cmd/proxy-loadgen ` -requests 1000 -concurrency 32 -timeout 10s ``` -命令输出 JSON 报告，包含成功/失败分类、固定内存的延迟分位上界、吞吐和 Go -运行时内存/GC 快照。它尚不包含长连接保持、Extract 的专用数据准备/结果校验， -也不构成 100,000 QPS 证明。 +命令输出 JSON 报告，包含成功/失败分类、`tunnelsEstablished`、固定内存的连接握手 +延迟分位上界、吞吐和 Go 运行时内存/GC 快照。它尚不包含 Extract 的专用数据准备/ +结果校验，也不构成 100,000 QPS 证明。 ## 关键配置与入口 diff --git a/cmd/proxy-loadgen/main.go b/cmd/proxy-loadgen/main.go index c5798a9..b372c34 100644 --- a/cmd/proxy-loadgen/main.go +++ b/cmd/proxy-loadgen/main.go @@ -29,14 +29,16 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io. flags := flag.NewFlagSet("proxy-loadgen", flag.ContinueOnError) flags.SetOutput(stderr) targetURL := flags.String("target", "", "HTTP or HTTPS target URL") - proxyURL := flags.String("proxy", "", "optional HTTP or HTTPS forward proxy URL") + proxyURL := flags.String("proxy", "", "optional HTTP or HTTPS forward proxy URL; connect requires HTTP") + scenario := flags.String("scenario", string(loadgen.ScenarioHTTP), "workload scenario: http or connect") method := flags.String("method", http.MethodGet, "HTTP method") body := flags.String("body", "", "UTF-8 request body") requests := flags.Int("requests", 0, "fixed request count; mutually exclusive with -duration") duration := flags.Duration("duration", 0, "time-boxed workload duration") rate := flags.Int("rate", 0, "maximum request starts per second for -duration; zero saturates workers") concurrency := flags.Int("concurrency", 64, "maximum concurrent requests") - timeout := flags.Duration("timeout", 10*time.Second, "per-request timeout") + timeout := flags.Duration("timeout", 10*time.Second, "HTTP request or CONNECT establishment timeout") + hold := flags.Duration("hold", 0, "CONNECT tunnel hold duration; required by -scenario connect") var headers headerValues flags.Var(&headers, "header", "repeatable HTTP header in Name: Value form") if err := flags.Parse(args); err != nil { @@ -55,8 +57,9 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io. return 2 } report, err := run(ctx, loadgen.Options{ - TargetURL: *targetURL, ProxyURL: *proxyURL, Method: *method, Headers: parsedHeaders, RequestBody: []byte(*body), + Scenario: loadgen.Scenario(*scenario), TargetURL: *targetURL, ProxyURL: *proxyURL, Method: *method, Headers: parsedHeaders, RequestBody: []byte(*body), Requests: *requests, Duration: *duration, Rate: *rate, Concurrency: *concurrency, RequestTimeout: *timeout, + TunnelHold: *hold, }) if err != nil { if errors.Is(err, loadgen.ErrInvalidOptions) { diff --git a/cmd/proxy-loadgen/main_test.go b/cmd/proxy-loadgen/main_test.go index 537354d..8ccefcc 100644 --- a/cmd/proxy-loadgen/main_test.go +++ b/cmd/proxy-loadgen/main_test.go @@ -38,3 +38,19 @@ func TestExecuteRejectsInvalidWorkload(t *testing.T) { t.Fatalf("execute() = %d stdout=%q stderr=%q", code, stdout.String(), stderr.String()) } } + +func TestExecutePassesCONNECTScenario(t *testing.T) { + var received loadgen.Options + var stdout, stderr bytes.Buffer + code := execute(context.Background(), []string{ + "-scenario", "connect", "-target", "https://target.example/health", "-proxy", "http://gateway.example:8080", + "-requests", "2", "-concurrency", "1", "-timeout", "2s", "-hold", "30s", + }, func(_ context.Context, options loadgen.Options) (loadgen.Report, error) { + received = options + return loadgen.Report{Requests: 2, Completed: 2, Succeeded: 2, Duration: time.Second}, nil + }, &stdout, &stderr) + if code != 0 || received.Scenario != loadgen.ScenarioConnect || received.TunnelHold != 30*time.Second || + received.ProxyURL != "http://gateway.example:8080" || stderr.Len() != 0 { + t.Fatalf("execute() = %d; options=%+v stderr=%q", code, received, stderr.String()) + } +} diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 5eb8e54..8417ff4 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -297,11 +297,12 @@ EGRESS 的出口身份响应解析支持固定上限的纯文本和常见 JSON I Controller 即可生效。Redis 任务存储现已扩展 BASIC/EGRESS/TARGET 的独立有界索引；路由目标 Profile 在启用 Routing 与 Upstream 的组合上才进入调度。 -补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 场景。固定请求数和固定时长两种 -模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小延迟直方图， -输出状态分类、吞吐和 Go 内存/GC 快照。通用 `method/header/body` 参数可覆盖 Distribution -提取 HTTP 请求；CONNECT 长连接、Extract 的专用数据准备与结果校验、故障注入以及代表性集群 -报告仍未实现。 +补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 与 CONNECT 长连接场景。固定请求数 +和固定时长两种模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小 +延迟直方图，输出状态分类、CONNECT 建立数、吞吐和 Go 内存/GC 快照。CONNECT 以原始 TCP +握手连接 HTTP Gateway，建连成功后按 `hold` 保持，且不透明读取隧道内容。通用 +`method/header/body` 参数可覆盖 Distribution 提取 HTTP 请求；Extract 的专用数据准备与结果 +校验、故障注入以及代表性集群报告仍未实现。 ## Task 12: Machine-readable Contracts diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 91c2738..3e0464c 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -100,7 +100,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 以下已有设计、接口或部署位置，但尚无端到端生产实现： -1. `proxy-loadgen` 已提供有界 HTTP 请求进程；`proxy-checker` 的 BASIC 任务进程已经完成， +1. `proxy-loadgen` 已提供有界 HTTP 与 HTTP Gateway CONNECT 长连接进程；`proxy-checker` 的 BASIC 任务进程已经完成， `proxy-controller` 已完成 Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成 HTTP/Metrics 与控制面 Session 装配，但 Provider 和业务指标链未闭环。 @@ -119,7 +119,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 原子归并、Controller Reducer 和 Observation 上报 RPC 已完成；EGRESS、TARGET 生产任务调度与 REMOVE 编排仍待实现。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 -9. CONNECT 长连接/Extract 场景、真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 +9. Extract 专用场景、真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 10. 将 reject/wait/direct 接入 Distribution 运行链，补齐 Sequential 持久化恢复、跨实例 CAS 和 disabled candidate 语义。 11. 补齐 Proxy Capacity 动态降容契约、Reservation 全生命周期观测；短 TTL diff --git a/docs/requirements/traceability.md b/docs/requirements/traceability.md index 59aab01..f3b9217 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/TARGET 共享任务运行态、`proxy-loadgen` 有界 HTTP 场景均已实现 | +| 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/CONNECT 长连接场景均已实现 | | 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 的编排待完成 | @@ -87,5 +87,5 @@ | OPS-001 | 配置校验后构建不可变快照并原子替换 | 8959-8999 | 100k 索引、版本/epoch 与并发 Apply/Acquire 测试 | | OPS-002 | 优雅停机停止新请求/Fetch，等待现有流量后超时关闭 | 8981-9000 | Provider Run 收敛与 `Handler.Shutdown` HTTP 排空、Hijacked CONNECT 超时关闭测试 | | OPS-003 | PostgreSQL 只保存管理修订、Upstream/Routing 状态、Admin 审计与 Outbox | 当前会话 | ADR-006、`adminstate` 公用契约和六表 Schema 边界测试；真实 PostgreSQL 契约待完成 | -| OBS-001 | 指标禁止 Proxy IP、session、Client、完整 URL 高基数标签 | 9001-9029 | Controller Prometheus/探针模块已实现且当前只暴露无业务标签的 Go/进程指标；低基数业务 Collector 与描述符测试待实现 | +| OBS-001 | 指标禁止 Proxy IP、session、Client、完整 URL 高基数标签 | 9001-9029 | Controller Prometheus/探针模块已实现；Checker Collector 仅暴露 `proxy_pool_checker_tasks_dispatched_total{level}` 与 `proxy_pool_checker_observations_total{level,result}`，并以注册表测试固定标签集。Provider、提取和容量业务指标待实现 | | TEST-001 | 覆盖对话中列出的 11 个关键并发与故障场景 | 9030-9082 | 测试清单；Redis 活动池由 Memory/Redis 公用契约覆盖，跨进程故障场景仍按清单推进 | diff --git a/internal/loadgen/http.go b/internal/loadgen/http.go index 29bae50..79e8222 100644 --- a/internal/loadgen/http.go +++ b/internal/loadgen/http.go @@ -3,11 +3,14 @@ package loadgen import ( + "bufio" "bytes" "context" + "encoding/base64" "errors" "io" "math/bits" + "net" "net/http" "net/url" "runtime" @@ -21,9 +24,17 @@ var ErrInvalidOptions = errors.New("invalid load generator options") const maximumResponseDrainBytes = 64 << 10 +type Scenario string + +const ( + ScenarioHTTP Scenario = "http" + ScenarioConnect Scenario = "connect" +) + // Options bounds one HTTP workload. Requests selects a fixed-size run; // otherwise Duration selects a time-boxed run and Rate caps its start rate. type Options struct { + Scenario Scenario TargetURL string ProxyURL string Method string @@ -34,22 +45,25 @@ type Options struct { Rate int Concurrency int RequestTimeout time.Duration + TunnelHold time.Duration } type Report struct { - StartedAt time.Time - Duration time.Duration - Requests uint64 - Completed uint64 - Succeeded uint64 - Failed uint64 - Status4xx uint64 - Status5xx uint64 - TimeoutErrors uint64 - RequestErrors uint64 - Throughput float64 - Latency LatencyReport - Runtime RuntimeReport + Scenario Scenario + StartedAt time.Time + Duration time.Duration + Requests uint64 + Completed uint64 + Succeeded uint64 + Failed uint64 + Status4xx uint64 + Status5xx uint64 + TimeoutErrors uint64 + RequestErrors uint64 + TunnelsEstablished uint64 + Throughput float64 + Latency LatencyReport + Runtime RuntimeReport } // LatencyReport uses logarithmic microsecond buckets. Percentiles are upper @@ -71,15 +85,16 @@ type RuntimeReport struct { } type counters struct { - requests atomic.Uint64 - completed atomic.Uint64 - succeeded atomic.Uint64 - failed atomic.Uint64 - status4xx atomic.Uint64 - status5xx atomic.Uint64 - timeoutErrors atomic.Uint64 - requestErrors atomic.Uint64 - latency latencyHistogram + requests atomic.Uint64 + completed atomic.Uint64 + succeeded atomic.Uint64 + failed atomic.Uint64 + status4xx atomic.Uint64 + status5xx atomic.Uint64 + timeoutErrors atomic.Uint64 + requestErrors atomic.Uint64 + tunnelsEstablished atomic.Uint64 + latency latencyHistogram } type latencyHistogram struct { @@ -90,30 +105,44 @@ func Run(ctx context.Context, options Options) (Report, error) { if ctx == nil { return Report{}, ErrInvalidOptions } - client, normalized, err := newClient(options) + normalized, err := normalizeOptions(options) if err != nil { return Report{}, err } - defer client.CloseIdleConnections() + stats := &counters{} + var execute func(context.Context) + if normalized.Scenario == ScenarioHTTP { + client, clientErr := newHTTPClient(normalized) + if clientErr != nil { + return Report{}, clientErr + } + defer client.CloseIdleConnections() + execute = func(requestCtx context.Context) { executeHTTP(requestCtx, client, normalized, stats) } + } else { + execute = func(requestCtx context.Context) { executeCONNECT(requestCtx, normalized, stats) } + } started := time.Now() - stats := &counters{} if normalized.Requests > 0 { - runFixed(ctx, client, normalized, stats) + runFixed(ctx, normalized, execute) if err := ctx.Err(); err != nil { - return report(started, stats), err + return report(started, normalized.Scenario, stats), err } } else { - runForDuration(ctx, client, normalized, stats) + runForDuration(ctx, normalized, execute) } - return report(started, stats), nil + return report(started, normalized.Scenario, stats), nil } -func newClient(options Options) (*http.Client, Options, error) { +func normalizeOptions(options Options) (Options, error) { normalized := options + normalized.Scenario = Scenario(strings.ToLower(strings.TrimSpace(string(normalized.Scenario)))) normalized.Method = strings.ToUpper(strings.TrimSpace(normalized.Method)) normalized.Headers = normalized.Headers.Clone() normalized.RequestBody = bytes.Clone(normalized.RequestBody) + if normalized.Scenario == "" { + normalized.Scenario = ScenarioHTTP + } if normalized.Method == "" { normalized.Method = http.MethodGet } @@ -121,37 +150,71 @@ func newClient(options Options) (*http.Client, Options, error) { normalized.Concurrency <= 0 || normalized.RequestTimeout <= 0 || normalized.Requests < 0 || normalized.Duration < 0 || normalized.Rate < 0 || (normalized.Requests == 0 && normalized.Duration <= 0) || (normalized.Requests > 0 && (normalized.Duration != 0 || normalized.Rate != 0)) { - return nil, Options{}, ErrInvalidOptions + return Options{}, ErrInvalidOptions } target, err := url.Parse(normalized.TargetURL) if err != nil || target.Scheme == "" || target.Host == "" || (target.Scheme != "http" && target.Scheme != "https") { - return nil, Options{}, ErrInvalidOptions + return Options{}, ErrInvalidOptions + } + if normalized.Scenario != ScenarioHTTP && normalized.Scenario != ScenarioConnect { + return Options{}, ErrInvalidOptions + } + if normalized.Scenario == ScenarioHTTP { + if normalized.TunnelHold != 0 { + return Options{}, ErrInvalidOptions + } + if _, err := parseProxyURL(normalized.ProxyURL, false); err != nil { + return Options{}, err + } + return normalized, nil + } + if normalized.ProxyURL == "" || normalized.TunnelHold <= 0 || len(normalized.RequestBody) != 0 || normalized.Method != http.MethodGet { + return Options{}, ErrInvalidOptions + } + if _, err := parseProxyURL(normalized.ProxyURL, true); err != nil { + return Options{}, err + } + return normalized, nil +} + +func newHTTPClient(options Options) (*http.Client, error) { + parsed, err := parseProxyURL(options.ProxyURL, false) + if err != nil { + return nil, err } var proxy func(*http.Request) (*url.URL, error) - if normalized.ProxyURL != "" { - if strings.TrimSpace(normalized.ProxyURL) != normalized.ProxyURL { - return nil, Options{}, ErrInvalidOptions - } - parsed, err := url.Parse(normalized.ProxyURL) - if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") { - return nil, Options{}, ErrInvalidOptions - } + if parsed != nil { proxy = http.ProxyURL(parsed) } transport := &http.Transport{ Proxy: proxy, ForceAttemptHTTP2: false, - MaxConnsPerHost: normalized.Concurrency, - MaxIdleConns: normalized.Concurrency, - MaxIdleConnsPerHost: normalized.Concurrency, + MaxConnsPerHost: options.Concurrency, + MaxIdleConns: options.Concurrency, + MaxIdleConnsPerHost: options.Concurrency, IdleConnTimeout: 30 * time.Second, - TLSHandshakeTimeout: normalized.RequestTimeout, - ResponseHeaderTimeout: normalized.RequestTimeout, + TLSHandshakeTimeout: options.RequestTimeout, + ResponseHeaderTimeout: options.RequestTimeout, } - return &http.Client{Transport: transport}, normalized, nil + return &http.Client{Transport: transport}, nil } -func runFixed(ctx context.Context, client *http.Client, options Options, stats *counters) { +func parseProxyURL(value string, plainHTTPOnly bool) (*url.URL, error) { + if value == "" { + return nil, nil + } + if strings.TrimSpace(value) != value { + return nil, ErrInvalidOptions + } + parsed, err := url.Parse(value) + if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") || + (plainHTTPOnly && parsed.Scheme != "http") { + return nil, ErrInvalidOptions + } + return parsed, nil +} + +func runFixed(ctx context.Context, options Options, execute func(context.Context)) { jobs := make(chan struct{}, min(options.Concurrency, options.Requests)) var workers sync.WaitGroup for range options.Concurrency { @@ -162,7 +225,7 @@ func runFixed(ctx context.Context, client *http.Client, options Options, stats * if ctx.Err() != nil { return } - executeOne(ctx, client, options, stats) + execute(ctx) } }() } @@ -179,7 +242,7 @@ func runFixed(ctx context.Context, client *http.Client, options Options, stats * workers.Wait() } -func runForDuration(ctx context.Context, client *http.Client, options Options, stats *counters) { +func runForDuration(ctx context.Context, options Options, execute func(context.Context)) { workloadContext, cancel := context.WithTimeout(ctx, options.Duration) defer cancel() if options.Rate == 0 { @@ -189,7 +252,7 @@ func runForDuration(ctx context.Context, client *http.Client, options Options, s go func() { defer workers.Done() for workloadContext.Err() == nil { - executeOne(workloadContext, client, options, stats) + execute(workloadContext) } }() } @@ -208,7 +271,7 @@ func runForDuration(ctx context.Context, client *http.Client, options Options, s case <-workloadContext.Done(): return case <-jobs: - executeOne(workloadContext, client, options, stats) + execute(workloadContext) } } }() @@ -244,7 +307,7 @@ func runForDuration(ctx context.Context, client *http.Client, options Options, s } } -func executeOne(ctx context.Context, client *http.Client, options Options, stats *counters) { +func executeHTTP(ctx context.Context, client *http.Client, options Options, stats *counters) { stats.requests.Add(1) started := time.Now() requestContext, cancel := context.WithTimeout(ctx, options.RequestTimeout) @@ -293,17 +356,136 @@ func executeOne(ctx context.Context, client *http.Client, options Options, stats } } -func report(started time.Time, stats *counters) Report { +func executeCONNECT(ctx context.Context, options Options, stats *counters) { + stats.requests.Add(1) + started := time.Now() + proxy, err := parseProxyURL(options.ProxyURL, true) + if err != nil { + recordRequestError(started, stats, false) + return + } + target, err := url.Parse(options.TargetURL) + if err != nil { + recordRequestError(started, stats, false) + return + } + authority := target.Host + if target.Port() == "" { + if target.Scheme == "https" { + authority = net.JoinHostPort(target.Hostname(), "443") + } else { + authority = net.JoinHostPort(target.Hostname(), "80") + } + } + + requestContext, cancel := context.WithTimeout(ctx, options.RequestTimeout) + defer cancel() + connection, err := (&net.Dialer{}).DialContext(requestContext, "tcp", proxy.Host) + if err != nil { + recordRequestError(started, stats, errors.Is(requestContext.Err(), context.DeadlineExceeded)) + return + } + defer connection.Close() + if err := connection.SetDeadline(time.Now().Add(options.RequestTimeout)); err != nil { + recordRequestError(started, stats, false) + return + } + if err := writeCONNECTRequest(connection, authority, options.Headers, proxy.User); err != nil { + recordRequestError(started, stats, errors.Is(requestContext.Err(), context.DeadlineExceeded)) + return + } + response, err := http.ReadResponse(bufio.NewReader(connection), &http.Request{Method: http.MethodConnect}) + latency := time.Since(started) + stats.latency.Record(latency) + if err != nil { + stats.failed.Add(1) + stats.completed.Add(1) + if errors.Is(requestContext.Err(), context.DeadlineExceeded) { + stats.timeoutErrors.Add(1) + } else { + stats.requestErrors.Add(1) + } + return + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + stats.failed.Add(1) + stats.completed.Add(1) + switch { + case response.StatusCode >= http.StatusInternalServerError: + stats.status5xx.Add(1) + case response.StatusCode >= http.StatusBadRequest: + stats.status4xx.Add(1) + default: + stats.requestErrors.Add(1) + } + return + } + if err := connection.SetDeadline(time.Time{}); err != nil { + stats.failed.Add(1) + stats.requestErrors.Add(1) + stats.completed.Add(1) + return + } + stats.succeeded.Add(1) + stats.tunnelsEstablished.Add(1) + waitForTunnelHold(ctx, options.TunnelHold) + stats.completed.Add(1) +} + +func writeCONNECTRequest(connection net.Conn, authority string, headers http.Header, user *url.Userinfo) error { + if connection == nil || authority == "" { + return ErrInvalidOptions + } + requestHeaders := headers.Clone() + requestHeaders.Del("Host") + if user != nil && requestHeaders.Get("Proxy-Authorization") == "" { + password, _ := user.Password() + encoded := base64.StdEncoding.EncodeToString([]byte(user.Username() + ":" + password)) + requestHeaders.Set("Proxy-Authorization", "Basic "+encoded) + } + var payload bytes.Buffer + _, _ = payload.WriteString("CONNECT " + authority + " HTTP/1.1\r\nHost: " + authority + "\r\n") + if err := requestHeaders.Write(&payload); err != nil { + return err + } + _, _ = payload.WriteString("\r\n") + _, err := io.Copy(connection, &payload) + return err +} + +func waitForTunnelHold(ctx context.Context, hold time.Duration) { + timer := time.NewTimer(hold) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + } +} + +func recordRequestError(started time.Time, stats *counters, timeout bool) { + stats.latency.Record(time.Since(started)) + stats.failed.Add(1) + stats.completed.Add(1) + if timeout { + stats.timeoutErrors.Add(1) + return + } + stats.requestErrors.Add(1) +} + +func report(started time.Time, scenario Scenario, stats *counters) Report { duration := time.Since(started) completed := stats.completed.Load() var memory runtime.MemStats runtime.ReadMemStats(&memory) result := Report{ - StartedAt: started.UTC(), Duration: duration, Requests: stats.requests.Load(), Completed: completed, + Scenario: scenario, StartedAt: started.UTC(), Duration: duration, Requests: stats.requests.Load(), Completed: completed, Succeeded: stats.succeeded.Load(), Failed: stats.failed.Load(), Status4xx: stats.status4xx.Load(), Status5xx: stats.status5xx.Load(), TimeoutErrors: stats.timeoutErrors.Load(), RequestErrors: stats.requestErrors.Load(), - Latency: stats.latency.Report(completed), - Runtime: RuntimeReport{NumCPU: runtime.NumCPU(), Goroutines: runtime.NumGoroutine(), AllocBytes: memory.Alloc, SysBytes: memory.Sys, GCCount: memory.NumGC}, + TunnelsEstablished: stats.tunnelsEstablished.Load(), + Latency: stats.latency.Report(completed), + Runtime: RuntimeReport{NumCPU: runtime.NumCPU(), Goroutines: runtime.NumGoroutine(), AllocBytes: memory.Alloc, SysBytes: memory.Sys, GCCount: memory.NumGC}, } if duration > 0 { result.Throughput = float64(completed) / duration.Seconds() diff --git a/internal/loadgen/http_test.go b/internal/loadgen/http_test.go index 64ef3b1..62c69d0 100644 --- a/internal/loadgen/http_test.go +++ b/internal/loadgen/http_test.go @@ -1,9 +1,12 @@ package loadgen import ( + "bufio" "context" "errors" + "fmt" "io" + "net" "net/http" "net/http/httptest" "sync/atomic" @@ -67,6 +70,81 @@ func TestRunTimeBoxedRateIsBounded(t *testing.T) { } } +func TestRunExecutesBoundedCONNECTTunnelWorkload(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen(): %v", err) + } + defer listener.Close() + + var tunnels atomic.Int64 + serveDone := make(chan struct{}) + go func() { + defer close(serveDone) + for range 3 { + connection, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + go func(connection net.Conn) { + defer connection.Close() + request, readErr := http.ReadRequest(bufio.NewReader(connection)) + if readErr != nil || request.Method != http.MethodConnect || request.Host != "target.example:443" { + t.Errorf("CONNECT request = %+v, error = %v", request, readErr) + return + } + tunnels.Add(1) + _, _ = fmt.Fprint(connection, "HTTP/1.1 200 Connection Established\r\n\r\n") + _, _ = io.Copy(io.Discard, connection) + }(connection) + } + }() + + report, err := Run(context.Background(), Options{ + Scenario: ScenarioConnect, TargetURL: "https://target.example/health", ProxyURL: "http://" + listener.Addr().String(), + Requests: 3, Concurrency: 2, RequestTimeout: time.Second, TunnelHold: 10 * time.Millisecond, + }) + if err != nil || report.Requests != 3 || report.Completed != 3 || report.Succeeded != 3 || report.Failed != 0 || + report.TunnelsEstablished != 3 || tunnels.Load() != 3 || report.Latency.P50UpperBound <= 0 { + t.Fatalf("Run() = (%+v, %v); CONNECT requests=%d", report, err, tunnels.Load()) + } + <-serveDone +} + +func TestRunClassifiesCONNECTProxyRejection(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen(): %v", err) + } + defer listener.Close() + go func() { + connection, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + defer connection.Close() + _, _ = http.ReadRequest(bufio.NewReader(connection)) + _, _ = fmt.Fprint(connection, "HTTP/1.1 407 Proxy Authentication Required\r\nContent-Length: 0\r\n\r\n") + }() + + report, err := Run(context.Background(), Options{ + Scenario: ScenarioConnect, TargetURL: "https://target.example/health", ProxyURL: "http://" + listener.Addr().String(), + Requests: 1, Concurrency: 1, RequestTimeout: time.Second, TunnelHold: time.Millisecond, + }) + if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.Status4xx != 1 || report.TunnelsEstablished != 0 { + t.Fatalf("Run() = (%+v, %v)", report, err) + } +} + +func TestRunRejectsCONNECTWithoutProxyOrHold(t *testing.T) { + _, err := Run(context.Background(), Options{ + Scenario: ScenarioConnect, TargetURL: "https://target.example/health", Requests: 1, Concurrency: 1, RequestTimeout: time.Second, + }) + if !errors.Is(err, ErrInvalidOptions) { + t.Fatalf("Run(connect without proxy or hold) error = %v, want ErrInvalidOptions", err) + } +} + func TestRunRejectsUnboundedWorkload(t *testing.T) { _, err := Run(context.Background(), Options{TargetURL: "http://127.0.0.1:8080", Concurrency: 1}) if !errors.Is(err, ErrInvalidOptions) {