fix(acceptance): 修复线上双站压测失真

将线上压力生成器拆分为宁波、香港宿主机上的 K3s 外进程，按全局序号分片并保守合并站点报告，避免操作者本机上行带宽成为容量瓶颈。\n\n为每次负载执行生成独立幂等键和 Run 级图片变体，阻止三轮验收重放旧任务或复用历史媒体缓存，并新增直接 OSS 物化一致性门禁。\n\n验证：Go 全量测试、go vet、迁移安全检查、bash -n、ShellCheck、manual-release-test 和双宿主机 linux/amd64 启动冒烟通过。
2026-08-01 01:02:07 +08:00
parent 9fd9c267d6
commit be6ce7f78a
5 changed files with 359 additions and 42 deletions
@@ -56,6 +56,9 @@ type options struct {
 	mixedDuration  time.Duration
 	imageRate      float64
 	videoRate      float64
+	shardIndex     int
+	shardCount     int
+	executionID    string
 }

 type report struct {
@@ -194,12 +197,18 @@ func parseOptions() (options, error) {
 	var mixedDuration time.Duration
 	var imageRate float64
 	var videoRate float64
+	var shardIndex int
+	var shardCount int
+	var executionID string
 	flag.StringVar(&profile, "profile", env("AI_GATEWAY_ACCEPTANCE_PROFILE", "simulated-all"), "simulated-all, Gemini profile, video profile, or real-canary")
 	flag.StringVar(&reportPath, "report", env("AI_GATEWAY_ACCEPTANCE_REPORT", ""), "secret-safe JSON report path")
 	flag.DurationVar(&timeout, "timeout", 45*time.Minute, "overall timeout")
 	flag.DurationVar(&mixedDuration, "duration", envDuration("AI_GATEWAY_ACCEPTANCE_MIXED_DURATION", 10*time.Minute), "mixed workload duration")
 	flag.Float64Var(&imageRate, "image-rate", envFloat("AI_GATEWAY_ACCEPTANCE_IMAGE_RATE", 0), "mixed image requests per second")
 	flag.Float64Var(&videoRate, "video-rate", envFloat("AI_GATEWAY_ACCEPTANCE_VIDEO_RATE", 0), "mixed video requests per second")
+	flag.IntVar(&shardIndex, "shard-index", 0, "zero-based distributed load shard index")
+	flag.IntVar(&shardCount, "shard-count", 1, "distributed load shard count")
+	flag.StringVar(&executionID, "execution-id", "", "unique workload execution identifier")
 	flag.Parse()
 	opts := options{
 		gateways:       splitCSV(os.Getenv("AI_GATEWAY_ACCEPTANCE_GATEWAYS")),
@@ -218,9 +227,21 @@ func parseOptions() (options, error) {
 		mixedDuration:  mixedDuration,
 		imageRate:      imageRate,
 		videoRate:      videoRate,
+		shardIndex:     shardIndex,
+		shardCount:     shardCount,
+		executionID:    strings.ToLower(strings.TrimSpace(executionID)),
 	}
-	if len(opts.gateways) != 2 {
-		return options{}, errors.New("AI_GATEWAY_ACCEPTANCE_GATEWAYS must contain exactly two API base URLs")
+	if opts.executionID == "" {
+		opts.executionID = opts.profile
+	}
+	if matched, _ := regexp.MatchString(`^[a-z0-9][a-z0-9-]{0,127}$`, opts.executionID); !matched {
+		return options{}, errors.New("execution-id must contain only lowercase letters, numbers, and hyphens")
+	}
+	if opts.shardCount < 1 || opts.shardCount > 32 || opts.shardIndex < 0 || opts.shardIndex >= opts.shardCount {
+		return options{}, fmt.Errorf("invalid load shard %d/%d", opts.shardIndex, opts.shardCount)
+	}
+	if len(opts.gateways) < 1 || len(opts.gateways) > 2 || (opts.shardCount == 1 && len(opts.gateways) != 2) {
+		return options{}, errors.New("AI_GATEWAY_ACCEPTANCE_GATEWAYS must contain two API base URLs, or one URL for a distributed shard")
 	}
 	for index := range opts.gateways {
 		opts.gateways[index] = strings.TrimRight(opts.gateways[index], "/")
@@ -262,6 +283,17 @@ func parseOptions() (options, error) {
 	return opts, nil
 }

+func (o options) shardRequestCount(total int) int {
+	if total <= o.shardIndex {
+		return 0
+	}
+	return (total-1-o.shardIndex)/o.shardCount + 1
+}
+
+func (o options) logicalRequestIndex(localIndex int) int {
+	return o.shardIndex + localIndex*o.shardCount
+}
+
 func run(ctx context.Context, opts options) ([]phaseResult, error) {
 	var tlsConfig *tls.Config
 	if opts.gatewayTLSName != "" || opts.gatewayCAFile != "" {
@@ -289,23 +321,26 @@ func run(ctx context.Context, opts options) ([]phaseResult, error) {
 	runPhase := func(name string) error {
 		var result phaseResult
 		if profile, ok := acceptanceworkload.GeminiProfileByName(name); ok {
-			result = runGemini(ctx, client, opts, name, profile.Requests, profile.InputBytes, profile.OutputBytes, false)
+			result = runGemini(
+				ctx, client, opts, name, opts.shardRequestCount(profile.Requests),
+				profile.InputBytes, profile.OutputBytes, false, opts.shardIndex, opts.shardCount,
+			)
 		} else {
 			switch name {
 			case "smoke-gemini":
-				result = runGemini(ctx, client, opts, name, 4, 256<<10, 256<<10, false)
+				result = runGemini(ctx, client, opts, name, opts.shardRequestCount(4), 256<<10, 256<<10, false, opts.shardIndex, opts.shardCount)
 			case "smoke-video":
-				result = runVideo(ctx, client, opts, name, 10, false, opts.emulatorFixtureURLs(), false, 0)
+				result = runVideo(ctx, client, opts, name, opts.shardRequestCount(10), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
 			case "video-throughput":
-				result = runVideo(ctx, client, opts, name, 1200, false, opts.emulatorFixtureURLs(), false, 0)
+				result = runVideo(ctx, client, opts, name, opts.shardRequestCount(1200), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
 			case "video-recovery":
-				result = runVideo(ctx, client, opts, name, 96, true, opts.emulatorFixtureURLs(), false, 0)
+				result = runVideo(ctx, client, opts, name, opts.shardRequestCount(96), true, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
 			case "mixed-soak", "mixed-overload":
 				result = runMixed(ctx, client, opts, name == "mixed-overload")
 			case "real-gemini-canary":
-				result = runGemini(ctx, client, opts, name, 1, 256<<10, 0, true)
+				result = runGemini(ctx, client, opts, name, 1, 256<<10, 0, true, 0, 1)
 			case "real-video-canary":
-				result = runVideo(ctx, client, opts, name, 1, false, opts.realImageURLs, true, 0)
+				result = runVideo(ctx, client, opts, name, 1, false, opts.realImageURLs, true, 0, 1)
 			default:
 				return fmt.Errorf("unknown phase %q", name)
 			}
@@ -338,7 +373,12 @@ func runGemini(
 	inputBytes int,
 	expectedOutputBytes int,
 	realUpstream bool,
+	requestOffset int,
+	requestStride int,
 ) phaseResult {
+	if requestStride < 1 {
+		requestStride = 1
+	}
 	startedAt := time.Now()
 	expectedHash := ""
 	if expectedOutputBytes > 0 {
@@ -358,6 +398,7 @@ func runGemini(
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			logicalIndex := requestOffset + index*requestStride
 			select {
 			case slots <- struct{}{}:
 			case <-ctx.Done():
@@ -365,8 +406,8 @@ func runGemini(
 			}
 			defer func() { <-slots }()
 			requestStarted := time.Now()
-			endpoint := opts.gateways[index%len(opts.gateways)] + "/v1beta/models/" + url.PathEscape(opts.geminiModel) + ":generateContent"
-			requestBody := streamGeminiRequestBody(paddedPNGVariant(inputBytes, index+1))
+			endpoint := opts.gateways[logicalIndex%len(opts.gateways)] + "/v1beta/models/" + url.PathEscape(opts.geminiModel) + ":generateContent"
+			requestBody := streamGeminiRequestBody(paddedPNGVariant(inputBytes, geminiInputVariant(opts.runID+"/"+opts.executionID, logicalIndex)))
 			req, err := http.NewRequestWithContext(
 				ctx,
 				http.MethodPost,
@@ -377,7 +418,7 @@ func runGemini(
 				_ = requestBody.Close()
 			}
 			if err == nil {
-				opts.setHeaders(req, index, realUpstream)
+				opts.setHeaders(req, logicalIndex, realUpstream)
 				req.Header.Set("Content-Type", "application/json")
 				var response *http.Response
 				response, err = client.Do(req)
@@ -472,7 +513,11 @@ func runVideo(
 	imageURLs []string,
 	realUpstream bool,
 	requestOffset int,
+	requestStride int,
 ) phaseResult {
+	if requestStride < 1 {
+		requestStride = 1
+	}
 	startedAt := time.Now()
 	combinationCount := 128
 	if realUpstream {
@@ -494,7 +539,7 @@ func runVideo(
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			logicalIndex := index + requestOffset
+			logicalIndex := requestOffset + index*requestStride
 			requestStarted := time.Now()
 			startedTasks[index] = requestStarted
 			imageCount := acceptanceworkload.VideoImageCount(logicalIndex)
@@ -590,7 +635,7 @@ func runVideo(
 			continue
 		}
 		index, taskID := index, taskID
-		logicalIndex := index + requestOffset
+		logicalIndex := requestOffset + index*requestStride
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
@@ -685,7 +730,7 @@ func runMixed(
 				generating = false
 				continue
 			}
-			requestIndex := requests
+			requestIndex := opts.logicalRequestIndex(requests)
 			requests++
 			imageAccumulator += opts.imageRate
 			isImage := imageAccumulator >= totalRate
@@ -698,7 +743,7 @@ func runMixed(
 				defer func() { <-slots }()
 				var result phaseResult
 				if isImage {
-					result = runGemini(runCtx, client, opts, "mixed-image", 1, 256<<10, 256<<10, false)
+					result = runGemini(runCtx, client, opts, "mixed-image", 1, 256<<10, 256<<10, false, requestIndex, 1)
 				} else {
 					result = runVideo(
 						runCtx,
@@ -710,6 +755,7 @@ func runMixed(
 						opts.emulatorFixtureURLs(),
 						false,
 						requestIndex,
+						1,
 					)
 				}
 				latency := time.Since(startedAt)
@@ -940,6 +986,9 @@ func (o options) setHeaders(request *http.Request, requestIndex int, realUpstrea
 	request.Header.Set("Authorization", "Bearer "+o.apiKeys[requestIndex%len(o.apiKeys)])
 	request.Header.Set(runHeader, o.runID)
 	request.Header.Set(tokenHeader, o.runToken)
+	if request.Method == http.MethodPost {
+		request.Header.Set("Idempotency-Key", fmt.Sprintf("acceptance-%s-%s-%d", o.runID, o.executionID, requestIndex))
+	}
 	if realUpstream {
 		request.Header.Set(upstreamHeader, "real")
 	}
@@ -1115,6 +1164,11 @@ func paddedPNG(size int) []byte {
 	return paddedPNGVariant(size, 0)
 }

+func geminiInputVariant(runID string, logicalIndex int) int {
+	digest := sha256.Sum256([]byte(fmt.Sprintf("%s:%d", strings.TrimSpace(runID), logicalIndex)))
+	return int(binary.BigEndian.Uint64(digest[:8]) & uint64(^uint(0)>>1))
+}
+
 func paddedPNGVariant(size int, variant int) []byte {
 	base, _ := base64.StdEncoding.DecodeString("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=")
 	if size <= len(base) {