From d9d3a3d6c08741470bb00b75b86f52c5b965006a Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 7 Aug 2026 16:00:09 +0800 Subject: [PATCH] fix: account for queued loadgen rate starts --- README.md | 3 ++- deploy/kubernetes/base/configmap.yaml | 10 ++++++++++ internal/config/config_test.go | 19 +++++++++++++++++++ internal/loadgen/http.go | 9 +++++++-- internal/loadgen/http_test.go | 17 ++++++++++++++++- 5 files changed, 54 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index dcd525e..8656d04 100644 --- a/README.md +++ b/README.md @@ -263,7 +263,8 @@ go run ./cmd/proxy-loadgen ` 连接握手延迟分位上界、吞吐和 Go 运行时内存/GC 快照。它不会输出提取响应中的地址或 凭据，也不构成 100,000 QPS 证明。限速场景以 `Requests` 表示实际发起数；当 `RateStartsDropped` 非零时，目标速率受压测端并发容量或时间窗限制，报告吞吐不得 -标注为已达到配置的 `-rate`。 +标注为已达到配置的 `-rate`。每个生成的限速令牌均会计入实际发起或丢弃，因此 +`RateStartsGenerated = Requests + RateStartsDropped`。 需要保留单次容量证据时，使用 `-output REPORT_FILE` 同时写出 JSON 文件；文件在 同目录完整写入后才替换目标，标准输出仍保留相同报告，便于交给日志或指标系统。 diff --git a/deploy/kubernetes/base/configmap.yaml b/deploy/kubernetes/base/configmap.yaml index 5179992..48433d9 100644 --- a/deploy/kubernetes/base/configmap.yaml +++ b/deploy/kubernetes/base/configmap.yaml @@ -21,6 +21,16 @@ data: limits: maxConcurrentConnections: 100000 requestsPerMinutePerClient: 60000 + transport: + dialTimeout: 10s + handshakeTimeout: 15s + responseHeaderTimeout: 30s + idleConnTimeout: 90s + maxIdleConns: 20000 + maxIdleConnsPerHost: 32 + maxConnsPerHost: 32 + tunnelBufferBytes: 32768 + tunnelIdleTimeout: 5m retry: maxAttempts: 2 retryMethods: [GET, HEAD] diff --git a/internal/config/config_test.go b/internal/config/config_test.go index df50d24..9fa241f 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -204,6 +204,7 @@ func TestShippedDeploymentConfigurationsResolveEnvironment(t *testing.T) { wantGatewayUser string wantPostgresURL string wantRedisURL string + wantTransport GatewayTransport }{ { name: "local", @@ -211,6 +212,12 @@ func TestShippedDeploymentConfigurationsResolveEnvironment(t *testing.T) { wantGatewayUser: "local-gateway", wantPostgresURL: "postgres://proxy_pool:local-only-change-me@postgres:5432/proxy_pool?sslmode=disable", wantRedisURL: "redis://redis:6379/0", + wantTransport: GatewayTransport{ + MaxIdleConns: 20000, + MaxIdleConnsPerHost: 32, + MaxConnsPerHost: 32, + TunnelBufferBytes: 32768, + }, }, { name: "kubernetes", @@ -219,6 +226,17 @@ func TestShippedDeploymentConfigurationsResolveEnvironment(t *testing.T) { wantGatewayUser: "resolved-gateway-user", wantPostgresURL: "postgres://resolved", wantRedisURL: "redis://resolved", + wantTransport: GatewayTransport{ + DialTimeout: Duration(10 * time.Second), + HandshakeTimeout: Duration(15 * time.Second), + ResponseHeaderTimeout: Duration(30 * time.Second), + IdleConnTimeout: Duration(90 * time.Second), + MaxIdleConns: 20000, + MaxIdleConnsPerHost: 32, + MaxConnsPerHost: 32, + TunnelBufferBytes: 32768, + TunnelIdleTimeout: Duration(5 * time.Minute), + }, }, } for _, test := range tests { @@ -248,6 +266,7 @@ func TestShippedDeploymentConfigurationsResolveEnvironment(t *testing.T) { cfg.Admin.Auth.Token != "resolved-admin-token" || cfg.Storage.PostgresURL != test.wantPostgresURL || cfg.Storage.RedisURL != test.wantRedisURL || + cfg.Gateway.Transport != test.wantTransport || cfg.Upstreams["provider-a"].API.Auth.Value != "resolved-provider-a-token" || cfg.Upstreams["provider-b"].API.Auth.Value != "resolved-provider-b-token" { t.Fatalf("deployment values were not resolved: %+v", cfg.Redacted()) diff --git a/internal/loadgen/http.go b/internal/loadgen/http.go index ae57b88..11325da 100644 --- a/internal/loadgen/http.go +++ b/internal/loadgen/http.go @@ -331,7 +331,7 @@ func runForDuration(ctx context.Context, options Options, execute func(context.C for { select { case <-workloadContext.Done(): - workers.Wait() + waitForRateWorkers(&workers, jobs, stats) return case now := <-ticker.C: credit += float64(options.Rate) * now.Sub(last).Seconds() @@ -352,7 +352,7 @@ func runForDuration(ctx context.Context, options Options, execute func(context.C case jobs <- struct{}{}: case <-workloadContext.Done(): stats.rateStartsDropped.Add(uint64(tokens - index)) - workers.Wait() + waitForRateWorkers(&workers, jobs, stats) return } } @@ -360,6 +360,11 @@ func runForDuration(ctx context.Context, options Options, execute func(context.C } } +func waitForRateWorkers(workers *sync.WaitGroup, jobs <-chan struct{}, stats *counters) { + workers.Wait() + stats.rateStartsDropped.Add(uint64(len(jobs))) +} + type requestBuilder func(context.Context) (*http.Request, responseValidator, error) type responseValidator func(*http.Response) error diff --git a/internal/loadgen/http_test.go b/internal/loadgen/http_test.go index d6d88a4..abc5bf1 100644 --- a/internal/loadgen/http_test.go +++ b/internal/loadgen/http_test.go @@ -66,11 +66,26 @@ func TestRunTimeBoxedRateIsBounded(t *testing.T) { TargetURL: server.URL, Duration: 100 * time.Millisecond, Rate: 100, Concurrency: 2, RequestTimeout: time.Second, }) if err != nil || report.Requests == 0 || report.Requests > 20 || report.Requests != report.Completed || - report.Succeeded != report.Completed || requests.Load() != int64(report.Requests) { + report.Succeeded+report.Failed != report.Completed || report.RateStartsGenerated != report.Requests+report.RateStartsDropped || + requests.Load() > int64(report.Requests) { t.Fatalf("Run() = (%+v, %v); handler requests=%d", report, err, requests.Load()) } } +func TestRunForDurationAccountsQueuedRateStartsAtDeadline(t *testing.T) { + stats := &counters{} + runForDuration(context.Background(), Options{ + Duration: 40 * time.Millisecond, Rate: 1000, Concurrency: 1, + }, func(ctx context.Context) { + stats.requests.Add(1) + <-ctx.Done() + }, stats) + + if generated, accounted := stats.rateStartsGenerated.Load(), stats.requests.Load()+stats.rateStartsDropped.Load(); generated == 0 || generated != accounted { + t.Fatalf("rate starts generated=%d, accounted=%d", generated, accounted) + } +} + func TestRunReportsRateStartsDroppedByConcurrency(t *testing.T) { started := make(chan struct{}, 1) server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {