From 97a515ddf0664032edef72bacfb37a2ea2622e71 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 09:42:34 +0800 Subject: [PATCH] feat: validate extraction load workloads --- README.md | 22 +- cmd/proxy-loadgen/main.go | 9 +- cmd/proxy-loadgen/main_test.go | 16 ++ docs/development/implementation-plan.md | 8 +- docs/requirements/completion-audit.md | 5 +- docs/requirements/traceability.md | 2 +- internal/loadgen/http.go | 291 +++++++++++++++++++----- internal/loadgen/http_test.go | 79 +++++++ 8 files changed, 358 insertions(+), 74 deletions(-) diff --git a/README.md b/README.md index dba91e9..c432d74 100644 --- a/README.md +++ b/README.md @@ -198,7 +198,8 @@ go run ./cmd/proxy-loadgen ` ``` 使用 `-duration 30s -rate 5000` 可运行限速场景；省略 `-rate` 时固定数量 worker -会饱和发送。`-method`、重复的 `-header` 与 `-body` 可组合用于 Distribution 的提取接口： +会饱和发送。普通 HTTP 场景中，`-method`、重复的 `-header` 与 `-body` 可组合用于 +Distribution 的提取接口： ```powershell go run ./cmd/proxy-loadgen ` @@ -208,9 +209,22 @@ go run ./cmd/proxy-loadgen ` -requests 1000 -concurrency 32 -timeout 10s ``` -命令输出 JSON 报告，包含成功/失败分类、`tunnelsEstablished`、固定内存的连接握手 -延迟分位上界、吞吐和 Go 运行时内存/GC 快照。它尚不包含 Extract 的专用数据准备/ -结果校验，也不构成 100,000 QPS 证明。 +`extract` 场景则自动构造 POST 请求、每请求独立的 `X-Request-ID` 与 +`Idempotency-Key`，并只校验响应的 `requestId`、数量和同响应内代理 ID 唯一性： + +```powershell +go run ./cmd/proxy-loadgen ` + -scenario extract ` + -target http://CONTROLLER_HOST:8081/api/v1/proxies/extract ` + -header "X-API-Key: DISTRIBUTION_API_KEY" ` + -requests 1000 -concurrency 32 -timeout 10s ` + -extract-count 1 -extract-fulfillment partial +``` + +命令输出 JSON 报告，包含成功/失败分类、`TunnelsEstablished`、 +`ExtractResponsesValidated`、`ExtractReturned`、`ExtractValidationFailures`、固定内存的 +连接握手延迟分位上界、吞吐和 Go 运行时内存/GC 快照。它不会输出提取响应中的地址或 +凭据，也不构成 100,000 QPS 证明。 ## 关键配置与入口 diff --git a/cmd/proxy-loadgen/main.go b/cmd/proxy-loadgen/main.go index b372c34..a8dd231 100644 --- a/cmd/proxy-loadgen/main.go +++ b/cmd/proxy-loadgen/main.go @@ -30,8 +30,8 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io. flags.SetOutput(stderr) targetURL := flags.String("target", "", "HTTP or HTTPS target 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") + scenario := flags.String("scenario", string(loadgen.ScenarioHTTP), "workload scenario: http, connect, or extract") + method := flags.String("method", "", "HTTP method; defaults to GET or POST for extract") 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") @@ -39,6 +39,8 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io. concurrency := flags.Int("concurrency", 64, "maximum concurrent requests") 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") + extractCount := flags.Int("extract-count", 1, "proxies requested by each extract scenario request") + extractFulfillment := flags.String("extract-fulfillment", "partial", "extract fulfillment: partial or allOrNothing") var headers headerValues flags.Var(&headers, "header", "repeatable HTTP header in Name: Value form") if err := flags.Parse(args); err != nil { @@ -59,7 +61,8 @@ func execute(ctx context.Context, args []string, run loadRun, stdout, stderr io. report, err := run(ctx, loadgen.Options{ 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, + TunnelHold: *hold, + ExtractCount: *extractCount, ExtractFulfillment: *extractFulfillment, }) 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 8ccefcc..4dfc688 100644 --- a/cmd/proxy-loadgen/main_test.go +++ b/cmd/proxy-loadgen/main_test.go @@ -54,3 +54,19 @@ func TestExecutePassesCONNECTScenario(t *testing.T) { t.Fatalf("execute() = %d; options=%+v stderr=%q", code, received, stderr.String()) } } + +func TestExecutePassesExtractScenario(t *testing.T) { + var received loadgen.Options + var stdout, stderr bytes.Buffer + code := execute(context.Background(), []string{ + "-scenario", "extract", "-target", "http://controller.example:8081/api/v1/proxies/extract", + "-requests", "2", "-concurrency", "1", "-timeout", "2s", "-extract-count", "3", "-extract-fulfillment", "allOrNothing", + }, 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.ScenarioExtract || received.Method != "" || received.ExtractCount != 3 || + received.ExtractFulfillment != "allOrNothing" || 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 8417ff4..090ecb6 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -299,10 +299,10 @@ Profile 在启用 Routing 与 Upstream 的组合上才进入调度。 补充进度（2026-08-02）：已新增 `proxy-loadgen` HTTP 与 CONNECT 长连接场景。固定请求数 和固定时长两种模式均通过固定 worker 数与有界派发通道执行，可选 QPS 限速；报告使用固定大小 -延迟直方图，输出状态分类、CONNECT 建立数、吞吐和 Go 内存/GC 快照。CONNECT 以原始 TCP -握手连接 HTTP Gateway，建连成功后按 `hold` 保持，且不透明读取隧道内容。通用 -`method/header/body` 参数可覆盖 Distribution 提取 HTTP 请求；Extract 的专用数据准备与结果 -校验、故障注入以及代表性集群报告仍未实现。 +延迟直方图，输出状态分类、CONNECT 建立数、Extract 校验数、吞吐和 Go 内存/GC 快照。CONNECT +以原始 TCP 握手连接 HTTP Gateway，建连成功后按 `hold` 保持，且不透明读取隧道内容。`extract` +场景会自动生成独立 Request/Idempotency 标识，校验返回数量与单响应 ID 唯一性，且不记录地址或 +凭据。故障注入以及代表性集群报告仍未实现。 ## Task 12: Machine-readable Contracts diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 3e0464c..b7b64b1 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -100,7 +100,8 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 以下已有设计、接口或部署位置，但尚无端到端生产实现： -1. `proxy-loadgen` 已提供有界 HTTP 与 HTTP Gateway CONNECT 长连接进程；`proxy-checker` 的 BASIC 任务进程已经完成， +1. `proxy-loadgen` 已提供有界 HTTP、HTTP Gateway CONNECT 长连接和 Distribution Extract + 响应校验场景；`proxy-checker` 的 BASIC 任务进程已经完成， `proxy-controller` 已完成 Admin/Distribution/Metrics 与 PostgreSQL/Redis 启动装配，`proxy-gateway` 已完成 HTTP/Metrics 与控制面 Session 装配，但 Provider 和业务指标链未闭环。 @@ -119,7 +120,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 原子归并、Controller Reducer 和 Observation 上报 RPC 已完成；EGRESS、TARGET 生产任务调度与 REMOVE 编排仍待实现。 8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 -9. Extract 专用场景、真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 +9. 真实 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 f3b9217..8167c31 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/CONNECT 长连接场景均已实现 | +| 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/Extract 校验场景均已实现 | | 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 index 79e8222..76be38a 100644 --- a/internal/loadgen/http.go +++ b/internal/loadgen/http.go @@ -6,7 +6,10 @@ import ( "bufio" "bytes" "context" + "crypto/rand" "encoding/base64" + "encoding/hex" + "encoding/json" "errors" "io" "math/bits" @@ -14,6 +17,7 @@ import ( "net/http" "net/url" "runtime" + "strconv" "strings" "sync" "sync/atomic" @@ -22,48 +26,57 @@ import ( var ErrInvalidOptions = errors.New("invalid load generator options") -const maximumResponseDrainBytes = 64 << 10 +const ( + maximumResponseDrainBytes = 64 << 10 + maximumExtractResponseBytes = 256 << 10 +) type Scenario string const ( ScenarioHTTP Scenario = "http" ScenarioConnect Scenario = "connect" + ScenarioExtract Scenario = "extract" ) // 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 - Headers http.Header - RequestBody []byte - Requests int - Duration time.Duration - Rate int - Concurrency int - RequestTimeout time.Duration - TunnelHold time.Duration + Scenario Scenario + TargetURL string + ProxyURL string + Method string + Headers http.Header + RequestBody []byte + Requests int + Duration time.Duration + Rate int + Concurrency int + RequestTimeout time.Duration + TunnelHold time.Duration + ExtractCount int + ExtractFulfillment string } type Report struct { - 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 + 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 + ExtractResponsesValidated uint64 + ExtractReturned uint64 + ExtractValidationFailures uint64 + Throughput float64 + Latency LatencyReport + Runtime RuntimeReport } // LatencyReport uses logarithmic microsecond buckets. Percentiles are upper @@ -85,16 +98,19 @@ 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 - tunnelsEstablished 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 + extractResponsesValidated atomic.Uint64 + extractReturned atomic.Uint64 + extractValidationFailures atomic.Uint64 + latency latencyHistogram } type latencyHistogram struct { @@ -111,13 +127,22 @@ func Run(ctx context.Context, options Options) (Report, error) { } stats := &counters{} var execute func(context.Context) - if normalized.Scenario == ScenarioHTTP { + if normalized.Scenario == ScenarioHTTP || normalized.Scenario == ScenarioExtract { client, clientErr := newHTTPClient(normalized) if clientErr != nil { return Report{}, clientErr } defer client.CloseIdleConnections() - execute = func(requestCtx context.Context) { executeHTTP(requestCtx, client, normalized, stats) } + if normalized.Scenario == ScenarioHTTP { + execute = func(requestCtx context.Context) { executeHTTP(requestCtx, client, normalized, stats) } + } else { + workloadID, workloadErr := newExtractWorkloadID() + if workloadErr != nil { + return Report{}, workloadErr + } + workload := &extractWorkload{id: workloadID, options: normalized, stats: stats} + execute = func(requestCtx context.Context) { workload.execute(requestCtx, client) } + } } else { execute = func(requestCtx context.Context) { executeCONNECT(requestCtx, normalized, stats) } } @@ -144,7 +169,11 @@ func normalizeOptions(options Options) (Options, error) { normalized.Scenario = ScenarioHTTP } if normalized.Method == "" { - normalized.Method = http.MethodGet + if normalized.Scenario == ScenarioExtract { + normalized.Method = http.MethodPost + } else { + 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 || @@ -156,7 +185,7 @@ func normalizeOptions(options Options) (Options, error) { if err != nil || target.Scheme == "" || target.Host == "" || (target.Scheme != "http" && target.Scheme != "https") { return Options{}, ErrInvalidOptions } - if normalized.Scenario != ScenarioHTTP && normalized.Scenario != ScenarioConnect { + if normalized.Scenario != ScenarioHTTP && normalized.Scenario != ScenarioConnect && normalized.Scenario != ScenarioExtract { return Options{}, ErrInvalidOptions } if normalized.Scenario == ScenarioHTTP { @@ -168,6 +197,20 @@ func normalizeOptions(options Options) (Options, error) { } return normalized, nil } + if normalized.Scenario == ScenarioExtract { + if normalized.ProxyURL != "" || normalized.TunnelHold != 0 || len(normalized.RequestBody) != 0 || + normalized.Method != http.MethodPost || normalized.ExtractCount <= 0 || normalized.ExtractCount > 64 || + (normalized.ExtractFulfillment != "" && strings.TrimSpace(normalized.ExtractFulfillment) != normalized.ExtractFulfillment) { + return Options{}, ErrInvalidOptions + } + if normalized.ExtractFulfillment == "" { + normalized.ExtractFulfillment = "partial" + } + if normalized.ExtractFulfillment != "partial" && normalized.ExtractFulfillment != "allOrNothing" { + return Options{}, ErrInvalidOptions + } + return normalized, nil + } if normalized.ProxyURL == "" || normalized.TunnelHold <= 0 || len(normalized.RequestBody) != 0 || normalized.Method != http.MethodGet { return Options{}, ErrInvalidOptions } @@ -307,31 +350,42 @@ func runForDuration(ctx context.Context, options Options, execute func(context.C } } +type requestBuilder func(context.Context) (*http.Request, responseValidator, error) + +type responseValidator func(*http.Response) error + func executeHTTP(ctx context.Context, client *http.Client, options Options, stats *counters) { + executeHTTPRequest(ctx, client, options.RequestTimeout, stats, func(requestContext context.Context) (*http.Request, responseValidator, error) { + request, err := http.NewRequestWithContext( + requestContext, + options.Method, + options.TargetURL, + bytes.NewReader(options.RequestBody), + ) + if err != nil { + return nil, nil, err + } + request.Header = options.Headers.Clone() + return request, discardResponse, nil + }) +} + +func executeHTTPRequest(ctx context.Context, client *http.Client, timeout time.Duration, stats *counters, build requestBuilder) { stats.requests.Add(1) started := time.Now() - requestContext, cancel := context.WithTimeout(ctx, options.RequestTimeout) + requestContext, cancel := context.WithTimeout(ctx, timeout) defer cancel() - request, err := http.NewRequestWithContext( - requestContext, - options.Method, - options.TargetURL, - bytes.NewReader(options.RequestBody), - ) + request, validate, err := build(requestContext) if err != nil { - stats.failed.Add(1) - stats.requestErrors.Add(1) - stats.completed.Add(1) - stats.latency.Record(time.Since(started)) + recordRequestError(started, stats, false) 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) + stats.completed.Add(1) if errors.Is(err, context.DeadlineExceeded) || errors.Is(requestContext.Err(), context.DeadlineExceeded) { stats.timeoutErrors.Add(1) } else { @@ -339,10 +393,18 @@ func executeHTTP(ctx context.Context, client *http.Client, options Options, stat } return } - _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, maximumResponseDrainBytes)) - _ = response.Body.Close() + defer response.Body.Close() switch { case response.StatusCode >= http.StatusOK && response.StatusCode < http.StatusMultipleChoices: + if validate == nil { + validate = discardResponse + } + if err := validate(response); err != nil { + stats.failed.Add(1) + stats.requestErrors.Add(1) + stats.completed.Add(1) + return + } stats.succeeded.Add(1) case response.StatusCode >= http.StatusInternalServerError: stats.failed.Add(1) @@ -354,6 +416,114 @@ func executeHTTP(ctx context.Context, client *http.Client, options Options, stat stats.failed.Add(1) stats.requestErrors.Add(1) } + stats.completed.Add(1) +} + +func discardResponse(response *http.Response) error { + if response == nil || response.Body == nil { + return ErrInvalidOptions + } + _, err := io.Copy(io.Discard, io.LimitReader(response.Body, maximumResponseDrainBytes)) + return err +} + +type extractWorkload struct { + id string + options Options + stats *counters + next atomic.Uint64 +} + +func newExtractWorkloadID() (string, error) { + var value [12]byte + if _, err := rand.Read(value[:]); err != nil { + return "", err + } + return "loadgen-" + hex.EncodeToString(value[:]), nil +} + +func (workload *extractWorkload) execute(ctx context.Context, client *http.Client) { + sequence := workload.next.Add(1) + requestID := workload.id + "-" + strconv.FormatUint(sequence, 10) + executeHTTPRequest(ctx, client, workload.options.RequestTimeout, workload.stats, func(requestContext context.Context) (*http.Request, responseValidator, error) { + body, err := json.Marshal(struct { + Count int `json:"count"` + Fulfillment string `json:"fulfillment"` + }{Count: workload.options.ExtractCount, Fulfillment: workload.options.ExtractFulfillment}) + if err != nil { + return nil, nil, err + } + request, err := http.NewRequestWithContext(requestContext, http.MethodPost, workload.options.TargetURL, bytes.NewReader(body)) + if err != nil { + return nil, nil, err + } + request.Header = workload.options.Headers.Clone() + if request.Header == nil { + request.Header = make(http.Header) + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Request-ID", requestID) + request.Header.Set("Idempotency-Key", requestID) + return request, func(response *http.Response) error { + return validateExtractResponse(response, requestID, workload.options, workload.stats) + }, nil + }) +} + +func validateExtractResponse(response *http.Response, requestID string, options Options, stats *counters) error { + if response == nil || response.Body == nil { + stats.extractValidationFailures.Add(1) + return ErrInvalidOptions + } + body, err := io.ReadAll(io.LimitReader(response.Body, maximumExtractResponseBytes+1)) + if err != nil || len(body) > maximumExtractResponseBytes { + stats.extractValidationFailures.Add(1) + if err != nil { + return err + } + return ErrInvalidOptions + } + var payload struct { + RequestID string `json:"requestId"` + Requested int `json:"requested"` + Returned int `json:"returned"` + Proxies json.RawMessage `json:"proxies"` + } + if err := json.Unmarshal(body, &payload); err != nil || payload.RequestID != requestID || payload.Requested != options.ExtractCount || + len(payload.Proxies) == 0 || bytes.Equal(bytes.TrimSpace(payload.Proxies), []byte("null")) || + (options.ExtractFulfillment == "allOrNothing" && payload.Returned != options.ExtractCount) { + stats.extractValidationFailures.Add(1) + if err != nil { + return err + } + return ErrInvalidOptions + } + var proxies []struct { + ID string `json:"id"` + } + if err := json.Unmarshal(payload.Proxies, &proxies); err != nil || payload.Returned < 0 || payload.Returned > options.ExtractCount || + payload.Returned != len(proxies) { + stats.extractValidationFailures.Add(1) + if err != nil { + return err + } + return ErrInvalidOptions + } + ids := make(map[string]struct{}, len(proxies)) + for _, proxy := range proxies { + if proxy.ID == "" { + stats.extractValidationFailures.Add(1) + return ErrInvalidOptions + } + if _, duplicate := ids[proxy.ID]; duplicate { + stats.extractValidationFailures.Add(1) + return ErrInvalidOptions + } + ids[proxy.ID] = struct{}{} + } + stats.extractResponsesValidated.Add(1) + stats.extractReturned.Add(uint64(payload.Returned)) + return nil } func executeCONNECT(ctx context.Context, options Options, stats *counters) { @@ -483,9 +653,10 @@ func report(started time.Time, scenario Scenario, stats *counters) Report { 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}, + TunnelsEstablished: stats.tunnelsEstablished.Load(), + ExtractResponsesValidated: stats.extractResponsesValidated.Load(), ExtractReturned: stats.extractReturned.Load(), + ExtractValidationFailures: stats.extractValidationFailures.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 62c69d0..eb5a0ca 100644 --- a/internal/loadgen/http_test.go +++ b/internal/loadgen/http_test.go @@ -3,6 +3,7 @@ package loadgen import ( "bufio" "context" + "encoding/json" "errors" "fmt" "io" @@ -136,6 +137,74 @@ func TestRunClassifiesCONNECTProxyRejection(t *testing.T) { } } +func TestRunExecutesAndValidatesExtractWorkload(t *testing.T) { + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodPost || request.Header.Get("Content-Type") != "application/json" || + request.Header.Get("X-Request-ID") == "" || request.Header.Get("Idempotency-Key") == "" { + t.Errorf("extract request = method=%s content-type=%q request-id=%q idempotency=%q", + request.Method, request.Header.Get("Content-Type"), request.Header.Get("X-Request-ID"), request.Header.Get("Idempotency-Key")) + } + var body struct { + Count int `json:"count"` + Fulfillment string `json:"fulfillment"` + } + if err := json.NewDecoder(request.Body).Decode(&body); err != nil || body.Count != 2 || body.Fulfillment != "partial" { + t.Errorf("extract body = %+v, error = %v", body, err) + } + requests.Add(1) + response.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"},{"id":"proxy-b"}]}`, + request.Header.Get("X-Request-ID")) + })) + defer server.Close() + + report, err := Run(context.Background(), Options{ + Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 2, Concurrency: 1, RequestTimeout: time.Second, + ExtractCount: 2, ExtractFulfillment: "partial", + }) + if err != nil || report.Requests != 2 || report.Completed != 2 || report.Succeeded != 2 || report.Failed != 0 || + report.ExtractResponsesValidated != 2 || report.ExtractReturned != 4 || report.ExtractValidationFailures != 0 || requests.Load() != 2 { + t.Fatalf("Run() = (%+v, %v); extract requests=%d", report, err, requests.Load()) + } +} + +func TestRunRejectsInvalidExtractResponse(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + response.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"}]}`, + request.Header.Get("X-Request-ID")) + })) + defer server.Close() + + report, err := Run(context.Background(), Options{ + Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second, + ExtractCount: 2, ExtractFulfillment: "partial", + }) + if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.ExtractResponsesValidated != 0 || + report.ExtractValidationFailures != 1 || report.ExtractReturned != 0 { + t.Fatalf("Run() = (%+v, %v)", report, err) + } +} + +func TestRunRejectsDuplicateProxyIDsInExtractResponse(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + response.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(response, `{"requestId":%q,"requested":2,"returned":2,"proxies":[{"id":"proxy-a"},{"id":"proxy-a"}]}`, + request.Header.Get("X-Request-ID")) + })) + defer server.Close() + + report, err := Run(context.Background(), Options{ + Scenario: ScenarioExtract, TargetURL: server.URL, Requests: 1, Concurrency: 1, RequestTimeout: time.Second, + ExtractCount: 2, ExtractFulfillment: "partial", + }) + if err != nil || report.Succeeded != 0 || report.Failed != 1 || report.ExtractResponsesValidated != 0 || + report.ExtractValidationFailures != 1 { + 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, @@ -145,6 +214,16 @@ func TestRunRejectsCONNECTWithoutProxyOrHold(t *testing.T) { } } +func TestRunRejectsInvalidExtractOptions(t *testing.T) { + _, err := Run(context.Background(), Options{ + Scenario: ScenarioExtract, TargetURL: "http://127.0.0.1:8081/api/v1/proxies/extract", Requests: 1, Concurrency: 1, + RequestTimeout: time.Second, ExtractCount: 0, + }) + if !errors.Is(err, ErrInvalidOptions) { + t.Fatalf("Run(invalid extract options) 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) {