feat: add load generator connect scenario
This commit is contained in:
youfak
parent
b35092ccdd
commit
d04b7413eb
20
README.md
20
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 证明。
|
||||
|
||||
## 关键配置与入口
|
||||
|
||||
|
||||
@ -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) {
|
||||
|
||||
@ -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())
|
||||
}
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 公用契约覆盖，跨进程故障场景仍按清单推进 |
|
||||
|
||||
@ -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,9 +45,11 @@ type Options struct {
|
||||
Rate int
|
||||
Concurrency int
|
||||
RequestTimeout time.Duration
|
||||
TunnelHold time.Duration
|
||||
}
|
||||
|
||||
type Report struct {
|
||||
Scenario Scenario
|
||||
StartedAt time.Time
|
||||
Duration time.Duration
|
||||
Requests uint64
|
||||
@ -47,6 +60,7 @@ type Report struct {
|
||||
Status5xx uint64
|
||||
TimeoutErrors uint64
|
||||
RequestErrors uint64
|
||||
TunnelsEstablished uint64
|
||||
Throughput float64
|
||||
Latency LatencyReport
|
||||
Runtime RuntimeReport
|
||||
@ -79,6 +93,7 @@ type counters struct {
|
||||
status5xx atomic.Uint64
|
||||
timeoutErrors atomic.Uint64
|
||||
requestErrors atomic.Uint64
|
||||
tunnelsEstablished atomic.Uint64
|
||||
latency latencyHistogram
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
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,15 +356,134 @@ 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(),
|
||||
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},
|
||||
}
|
||||
|
||||
@ -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) {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user