From 8fdeb0a9057ff657af98f1ba3dcd8d4a8aa3c6d7 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 08:10:45 +0800 Subject: [PATCH] feat: add bounded HTTP load generator --- README.md | 18 +- cmd/proxy-loadgen/main.go | 101 +++++++ cmd/proxy-loadgen/main_test.go | 40 +++ docs/design/architecture.md | 6 +- docs/design/project-structure.md | 7 +- docs/development/implementation-plan.md | 5 + docs/operations/runbook.md | 5 +- docs/requirements/completion-audit.md | 4 +- docs/requirements/traceability.md | 2 +- internal/loadgen/http.go | 339 ++++++++++++++++++++++++ internal/loadgen/http_test.go | 70 +++++ 11 files changed, 585 insertions(+), 12 deletions(-) create mode 100644 cmd/proxy-loadgen/main.go create mode 100644 cmd/proxy-loadgen/main_test.go create mode 100644 internal/loadgen/http.go create mode 100644 internal/loadgen/http_test.go diff --git a/README.md b/README.md index ce3c9e3..8cfc333 100644 --- a/README.md +++ b/README.md @@ -93,7 +93,7 @@ flowchart LR Observation 状态归并。 - **部分完成**：EGRESS 与 TARGET 的任务编排和配置建模，Docker Compose/Kubernetes 运行时 mTLS Overlay。 -- **待完成**：loadgen、故障演练和代表性集群压测。 +- **待完成**：CONNECT 长连接/Extract 压测场景、故障演练和代表性集群压测。 检查项数量不等于生产就绪度。静态部署清单与 protobuf descriptor 验证也不代表 端到端拓扑已经完成；`100,000 QPS` 仍只是待验证的集群设计目标。 @@ -171,7 +171,21 @@ Checker 的参数也可通过 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、 Controller 在启用控制面时装配 Redis 共享任务 broker，并按启用的 Upstream 调度 HTTP/HTTPS/SOCKS5 BASIC 检查。调度监督器每轮读取已发布配置，因此 reload 后的上游启停、 检查间隔、抖动、超时、重试次数和 `maxInFlight` 会在下一轮生效；新启用的上游无需 -重启 Controller。EGRESS 与 TARGET 尚未进入生产调度，loadgen 命令也尚未实现。 +重启 Controller。EGRESS 与 TARGET 尚未进入生产调度。 + +初版 HTTP 容量工具可按固定请求数或固定时长运行，并将 HTTPS 目标经 Gateway 的请求 +交给标准 HTTP Transport 建立 CONNECT： + +```powershell +go run ./cmd/proxy-loadgen ` + -target https://TARGET_URL/health ` + -proxy http://GATEWAY_HOST:8080 ` + -requests 10000 -concurrency 128 -timeout 10s +``` + +使用 `-duration 30s -rate 5000` 可运行限速场景；省略 `-rate` 时固定数量 worker +会饱和发送。命令输出 JSON 报告，包含成功/失败分类、固定内存的延迟分位上界、吞吐和 +Go 运行时内存/GC 快照。它不包含长连接保持或 Extract 场景，也不构成 100,000 QPS 证明。 ## 关键配置与入口 diff --git a/cmd/proxy-loadgen/main.go b/cmd/proxy-loadgen/main.go new file mode 100644 index 0000000..33f40d3 --- /dev/null +++ b/cmd/proxy-loadgen/main.go @@ -0,0 +1,101 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "proxy-pool/internal/loadgen" +) + +type loadRun func(context.Context, loadgen.Options) (loadgen.Report, error) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + os.Exit(execute(ctx, os.Args[1:], loadgen.Run, os.Stdout, os.Stderr)) +} + +func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io.Writer) int { + 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") + method := flags.String("method", http.MethodGet, "HTTP method") + 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") + var headers headerValues + flags.Var(&headers, "header", "repeatable HTTP header in Name: Value form") + if err := flags.Parse(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + return 0 + } + return 2 + } + if flags.NArg() != 0 || ctx == nil || run == nil { + _, _ = fmt.Fprintln(stderr, "proxy-loadgen: target and a bounded workload are required") + return 2 + } + parsedHeaders, err := headers.Header() + if err != nil { + _, _ = fmt.Fprintf(stderr, "proxy-loadgen: %v\n", err) + return 2 + } + report, err := run(ctx, loadgen.Options{ + TargetURL: *targetURL, ProxyURL: *proxyURL, Method: *method, Headers: parsedHeaders, + Requests: *requests, Duration: *duration, Rate: *rate, Concurrency: *concurrency, RequestTimeout: *timeout, + }) + if err != nil { + if errors.Is(err, loadgen.ErrInvalidOptions) { + _, _ = fmt.Fprintln(stderr, "proxy-loadgen: -target, positive -concurrency/-timeout, and exactly one bounded workload are required") + return 2 + } + _, _ = fmt.Fprintf(stderr, "proxy-loadgen: %v\n", err) + return 1 + } + if err := json.NewEncoder(stdout).Encode(report); err != nil { + _, _ = fmt.Fprintf(stderr, "proxy-loadgen: encode report: %v\n", err) + return 1 + } + return 0 +} + +type headerValues []string + +func (values *headerValues) String() string { + return strings.Join(*values, ",") +} + +func (values *headerValues) Set(value string) error { + if strings.TrimSpace(value) != value || value == "" { + return errors.New("invalid header") + } + *values = append(*values, value) + return nil +} + +func (values headerValues) Header() (http.Header, error) { + result := make(http.Header, len(values)) + for _, value := range values { + name, content, found := strings.Cut(value, ":") + name = strings.TrimSpace(name) + content = strings.TrimSpace(content) + if !found || name == "" || strings.ContainsAny(name, "\r\n") || strings.ContainsAny(content, "\r\n") { + return nil, errors.New("invalid header") + } + result.Add(name, content) + } + return result, nil +} diff --git a/cmd/proxy-loadgen/main_test.go b/cmd/proxy-loadgen/main_test.go new file mode 100644 index 0000000..acd3391 --- /dev/null +++ b/cmd/proxy-loadgen/main_test.go @@ -0,0 +1,40 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "testing" + "time" + + "proxy-pool/internal/loadgen" +) + +func TestExecutePassesBoundedWorkloadAndWritesJSON(t *testing.T) { + var received loadgen.Options + var stdout, stderr bytes.Buffer + code := execute(context.Background(), []string{ + "-target", "https://target.example/path", "-proxy", "http://gateway.example:8080", "-method", "post", + "-requests", "3", "-concurrency", "2", "-timeout", "2s", "-header", "X-Run: fixed", + }, func(_ context.Context, options loadgen.Options) (loadgen.Report, error) { + received = options + return loadgen.Report{Requests: 3, Completed: 3, Succeeded: 3, Duration: time.Second}, nil + }, &stdout, &stderr) + if code != 0 || received.TargetURL != "https://target.example/path" || received.ProxyURL != "http://gateway.example:8080" || + received.Method != "post" || received.Requests != 3 || received.Duration != 0 || received.Concurrency != 2 || + received.Headers.Get("X-Run") != "fixed" || stderr.Len() != 0 { + t.Fatalf("execute() = %d; options=%+v stderr=%q", code, received, stderr.String()) + } + var report loadgen.Report + if err := json.Unmarshal(stdout.Bytes(), &report); err != nil || report.Succeeded != 3 { + t.Fatalf("JSON report = (%+v, %v)", report, err) + } +} + +func TestExecuteRejectsInvalidWorkload(t *testing.T) { + var stdout, stderr bytes.Buffer + code := execute(context.Background(), []string{"-target", "http://target.example"}, loadgen.Run, &stdout, &stderr) + if code != 2 || stdout.Len() != 0 || stderr.Len() == 0 { + t.Fatalf("execute() = %d stdout=%q stderr=%q", code, stdout.String(), stderr.String()) + } +} diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 1ec159a..52071dd 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -79,8 +79,10 @@ Provider、Pool、Routing、Distribution 在首版需要共享事务和一致性 ### 3.4 proxy-loadgen -- 分别生成 HTTP QPS、CONNECT 活跃连接、建连速率和 Extract 并发。 -- 输出环境、场景、延迟、错误、CPU、RSS、句柄和网络结果。 +- 当前实现 HTTP 请求场景：固定请求数或固定时长，受限并发与可选目标 QPS；HTTPS + 目标经 HTTP Gateway 时由 Transport 走 CONNECT。 +- 输出场景、延迟分位上界、错误分类、吞吐和 Go 内存/GC 快照。CONNECT 长连接、 + Extract 并发、进程 CPU/RSS/句柄与网络采样仍需补齐。 ## 4. 模块边界 diff --git a/docs/design/project-structure.md b/docs/design/project-structure.md index 3f6e30d..f643d5a 100644 --- a/docs/design/project-structure.md +++ b/docs/design/project-structure.md @@ -53,9 +53,10 @@ deadline 内执行 HTTP/HTTPS/SOCKS5 BASIC、EGRESS 和 TARGET 探测并微批 ### proxy-loadgen -负载工具生成 HTTP、CONNECT、连接复用与故障注入场景，输出延迟分位数、 -错误类别、连接数、CPU、RSS、GC 与吞吐。它是 100k QPS 结论的证据工具， -不是业务进程。 +负载工具当前生成有界 HTTP 请求，支持经 Gateway 请求 HTTPS 目标、固定请求数或时长、 +目标 QPS、连接复用和 JSON 指标输出。延迟统计使用固定大小直方图，不会因长时间高 QPS +运行积压样本。CONNECT 长连接、Extract 和故障注入场景仍待补齐；它是 100k QPS 结论的 +证据工具，不是业务进程。 ## 3. 依赖方向 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index da85c10..f2fc9de 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -293,6 +293,11 @@ EGRESS/TARGET 多维任务索引及部署运行态仍未实现， 有界派发逻辑，所以 reload 后已启用上游的策略变更、停用，以及新启用上游都无需重启 Controller 即可生效；Redis 任务存储仍仅承载 BASIC，未扩展 EGRESS/TARGET 的多维索引。 +补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 场景。固定请求数和固定时长两种 +模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小延迟直方图， +输出状态分类、吞吐和 Go 内存/GC 快照。CONNECT 长连接、Extract、故障注入以及代表性集群 +报告仍未实现。 + ## Task 12: Machine-readable Contracts **Files:** `api/openapi/proxy-pool.yaml`, `api/proto/controlplane/v1/controlplane.proto`, diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index 95cc5bd..2477fda 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -24,8 +24,9 @@ `cmd/proxy-controller` 已完成配置单次加载、PostgreSQL 迁移、Redis 活动池、 Distribution/Admin/Metrics 独立监听和有界停机装配。Provider 自动补池、分布式 配额、动态重载和 Admin 低基数统计已装配；Controller 已装配 Redis BASIC 任务 broker， -`proxy-checker` 可执行 HTTP/HTTPS/SOCKS5 BASIC 探测。EGRESS/TARGET 调度、loadgen 与完整 -mTLS 环境 Overlay 仍属于 `implementation-plan.md` 后续任务。 +`proxy-checker` 可执行 HTTP/HTTPS/SOCKS5 BASIC 探测。`proxy-loadgen` 已提供有界 HTTP +请求场景；EGRESS/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 bb0b202..8ab1790 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -99,7 +99,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 以下已有设计、接口或部署位置，但尚无端到端生产实现： -1. `proxy-loadgen` 进程装配；`proxy-checker` 的 BASIC 任务进程已经完成， +1. `proxy-loadgen` 已提供有界 HTTP 请求进程；`proxy-checker` 的 BASIC 任务进程已经完成， `proxy-controller` 已完成 Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成 HTTP/Metrics 与控制面 Session 装配，但 Provider 和业务指标链未闭环。 @@ -118,7 +118,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 原子归并、Controller Reducer 和 Observation 上报 RPC 已完成；EGRESS、TARGET 生产任务调度与 REMOVE 编排仍待实现。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 -9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 +9. CONNECT 长连接/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 f319497..6845a25 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 共享任务运行态已实现；loadgen、EGRESS/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 共享任务运行态、`proxy-loadgen` 有界 HTTP 场景均已实现；EGRESS/TARGET 生产编排待实现 | | 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 的编排待完成 | diff --git a/internal/loadgen/http.go b/internal/loadgen/http.go new file mode 100644 index 0000000..443d323 --- /dev/null +++ b/internal/loadgen/http.go @@ -0,0 +1,339 @@ +// Package loadgen runs bounded, reproducible HTTP request workloads. It is a +// capacity-evidence tool and is deliberately separate from service processes. +package loadgen + +import ( + "context" + "errors" + "io" + "math/bits" + "net/http" + "net/url" + "runtime" + "strings" + "sync" + "sync/atomic" + "time" +) + +var ErrInvalidOptions = errors.New("invalid load generator options") + +const maximumResponseDrainBytes = 64 << 10 + +// 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 { + TargetURL string + ProxyURL string + Method string + Headers http.Header + Requests int + Duration time.Duration + Rate int + Concurrency int + RequestTimeout 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 +} + +// LatencyReport uses logarithmic microsecond buckets. Percentiles are upper +// bounds so the collector remains fixed-size even under high QPS. +type LatencyReport struct { + Samples uint64 + P50UpperBound time.Duration + P95UpperBound time.Duration + P99UpperBound time.Duration + MaximumUpperBound time.Duration +} + +type RuntimeReport struct { + NumCPU int + Goroutines int + AllocBytes uint64 + SysBytes uint64 + GCCount uint32 +} + +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 +} + +type latencyHistogram struct { + buckets [64]atomic.Uint64 +} + +func Run(ctx context.Context, options Options) (Report, error) { + if ctx == nil { + return Report{}, ErrInvalidOptions + } + client, normalized, err := newClient(options) + if err != nil { + return Report{}, err + } + defer client.CloseIdleConnections() + + started := time.Now() + stats := &counters{} + if normalized.Requests > 0 { + runFixed(ctx, client, normalized, stats) + if err := ctx.Err(); err != nil { + return report(started, stats), err + } + } else { + runForDuration(ctx, client, normalized, stats) + } + return report(started, stats), nil +} + +func newClient(options Options) (*http.Client, Options, error) { + normalized := options + normalized.Method = strings.ToUpper(strings.TrimSpace(normalized.Method)) + if normalized.Method == "" { + normalized.Method = http.MethodGet + } + if strings.TrimSpace(normalized.TargetURL) != normalized.TargetURL || normalized.TargetURL == "" || + 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 + } + target, err := url.Parse(normalized.TargetURL) + if err != nil || target.Scheme == "" || target.Host == "" || (target.Scheme != "http" && target.Scheme != "https") { + return nil, Options{}, ErrInvalidOptions + } + 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 + } + proxy = http.ProxyURL(parsed) + } + transport := &http.Transport{ + Proxy: proxy, + ForceAttemptHTTP2: false, + MaxConnsPerHost: normalized.Concurrency, + MaxIdleConns: normalized.Concurrency, + MaxIdleConnsPerHost: normalized.Concurrency, + IdleConnTimeout: 30 * time.Second, + TLSHandshakeTimeout: normalized.RequestTimeout, + ResponseHeaderTimeout: normalized.RequestTimeout, + } + return &http.Client{Transport: transport}, normalized, nil +} + +func runFixed(ctx context.Context, client *http.Client, options Options, stats *counters) { + jobs := make(chan struct{}, min(options.Concurrency, options.Requests)) + var workers sync.WaitGroup + for range options.Concurrency { + workers.Add(1) + go func() { + defer workers.Done() + for range jobs { + if ctx.Err() != nil { + return + } + executeOne(ctx, client, options, stats) + } + }() + } + for range options.Requests { + select { + case jobs <- struct{}{}: + case <-ctx.Done(): + close(jobs) + workers.Wait() + return + } + } + close(jobs) + workers.Wait() +} + +func runForDuration(ctx context.Context, client *http.Client, options Options, stats *counters) { + workloadContext, cancel := context.WithTimeout(ctx, options.Duration) + defer cancel() + if options.Rate == 0 { + var workers sync.WaitGroup + for range options.Concurrency { + workers.Add(1) + go func() { + defer workers.Done() + for workloadContext.Err() == nil { + executeOne(workloadContext, client, options, stats) + } + }() + } + workers.Wait() + return + } + + jobs := make(chan struct{}, options.Concurrency) + var workers sync.WaitGroup + for range options.Concurrency { + workers.Add(1) + go func() { + defer workers.Done() + for { + select { + case <-workloadContext.Done(): + return + case <-jobs: + executeOne(workloadContext, client, options, stats) + } + } + }() + } + + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + last := time.Now() + credit := 0.0 + for { + select { + case <-workloadContext.Done(): + workers.Wait() + return + case now := <-ticker.C: + credit += float64(options.Rate) * now.Sub(last).Seconds() + last = now + tokens := int(credit) + credit -= float64(tokens) + if tokens > options.Concurrency { + tokens = options.Concurrency + credit = 0 + } + for range tokens { + select { + case jobs <- struct{}{}: + case <-workloadContext.Done(): + workers.Wait() + return + } + } + } + } +} + +func executeOne(ctx context.Context, client *http.Client, options Options, stats *counters) { + stats.requests.Add(1) + started := time.Now() + requestContext, cancel := context.WithTimeout(ctx, options.RequestTimeout) + defer cancel() + request, err := http.NewRequestWithContext(requestContext, options.Method, options.TargetURL, nil) + if err != nil { + stats.failed.Add(1) + stats.requestErrors.Add(1) + stats.completed.Add(1) + stats.latency.Record(time.Since(started)) + return + } + request.Header = options.Headers.Clone() + response, err := client.Do(request) + latency := time.Since(started) + stats.latency.Record(latency) + stats.completed.Add(1) + if err != nil { + stats.failed.Add(1) + if errors.Is(err, context.DeadlineExceeded) || errors.Is(requestContext.Err(), context.DeadlineExceeded) { + stats.timeoutErrors.Add(1) + } else { + stats.requestErrors.Add(1) + } + return + } + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, maximumResponseDrainBytes)) + _ = response.Body.Close() + switch { + case response.StatusCode >= http.StatusOK && response.StatusCode < http.StatusMultipleChoices: + stats.succeeded.Add(1) + case response.StatusCode >= http.StatusInternalServerError: + stats.failed.Add(1) + stats.status5xx.Add(1) + case response.StatusCode >= http.StatusBadRequest: + stats.failed.Add(1) + stats.status4xx.Add(1) + default: + stats.failed.Add(1) + stats.requestErrors.Add(1) + } +} + +func report(started time.Time, 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, + 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}, + } + if duration > 0 { + result.Throughput = float64(completed) / duration.Seconds() + } + return result +} + +func (histogram *latencyHistogram) Record(latency time.Duration) { + microseconds := latency.Microseconds() + if microseconds < 1 { + microseconds = 1 + } + index := bits.Len64(uint64(microseconds - 1)) + if index >= len(histogram.buckets) { + index = len(histogram.buckets) - 1 + } + histogram.buckets[index].Add(1) +} + +func (histogram *latencyHistogram) Report(samples uint64) LatencyReport { + result := LatencyReport{Samples: samples} + if samples == 0 { + return result + } + result.P50UpperBound = histogram.percentile(samples, 50) + result.P95UpperBound = histogram.percentile(samples, 95) + result.P99UpperBound = histogram.percentile(samples, 99) + result.MaximumUpperBound = histogram.percentile(samples, 100) + return result +} + +func (histogram *latencyHistogram) percentile(samples uint64, percentile uint64) time.Duration { + target := (samples*percentile + 99) / 100 + seen := uint64(0) + for index := range histogram.buckets { + seen += histogram.buckets[index].Load() + if seen >= target { + return time.Duration(uint64(1)< 20 || report.Requests != report.Completed || + report.Succeeded != report.Completed || requests.Load() != int64(report.Requests) { + t.Fatalf("Run() = (%+v, %v); handler requests=%d", report, err, requests.Load()) + } +} + +func TestRunRejectsUnboundedWorkload(t *testing.T) { + _, err := Run(context.Background(), Options{TargetURL: "http://127.0.0.1:8080", Concurrency: 1}) + if !errors.Is(err, ErrInvalidOptions) { + t.Fatalf("Run(unbounded) error = %v, want ErrInvalidOptions", err) + } +}