diff --git a/apps/api/cmd/acceptance-load/main.go b/apps/api/cmd/acceptance-load/main.go index 78060bc..d124487 100644 --- a/apps/api/cmd/acceptance-load/main.go +++ b/apps/api/cmd/acceptance-load/main.go @@ -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) { diff --git a/apps/api/cmd/acceptance-load/main_test.go b/apps/api/cmd/acceptance-load/main_test.go index 0df750e..2564f38 100644 --- a/apps/api/cmd/acceptance-load/main_test.go +++ b/apps/api/cmd/acceptance-load/main_test.go @@ -146,7 +146,7 @@ func TestGeminiLoadIsSplitAcrossTwoGatewayAPIs(t *testing.T) { gateways: []string{first.URL, second.URL}, apiKeys: []string{"key-1", "key-2"}, runID: "run-1", runToken: "token-1", geminiModel: "gemini-image-test", } - result := runGemini(t.Context(), http.DefaultClient, opts, "dual-api", 8, 256<<10, 256<<10, false) + result := runGemini(t.Context(), http.DefaultClient, opts, "dual-api", 8, 256<<10, 256<<10, false, 0, 1) if result.err != nil || result.report.Completed != 8 { t.Fatalf("result=%+v err=%v", result.report, result.err) } @@ -158,6 +158,53 @@ func TestGeminiLoadIsSplitAcrossTwoGatewayAPIs(t *testing.T) { } } +func TestDistributedShardIndexesCoverWorkloadWithoutOverlap(t *testing.T) { + first := options{shardIndex: 0, shardCount: 2} + second := options{shardIndex: 1, shardCount: 2} + if first.shardRequestCount(1001) != 501 || second.shardRequestCount(1001) != 500 { + t.Fatalf("shard counts=%d/%d", first.shardRequestCount(1001), second.shardRequestCount(1001)) + } + seen := map[int]bool{} + for local := 0; local < first.shardRequestCount(1001); local++ { + seen[first.logicalRequestIndex(local)] = true + } + for local := 0; local < second.shardRequestCount(1001); local++ { + index := second.logicalRequestIndex(local) + if seen[index] { + t.Fatalf("duplicate logical index %d", index) + } + seen[index] = true + } + if len(seen) != 1001 { + t.Fatalf("covered indexes=%d", len(seen)) + } +} + +func TestGeminiInputVariantIsRunScoped(t *testing.T) { + if geminiInputVariant("run-a", 7) == geminiInputVariant("run-b", 7) { + t.Fatal("different Run IDs produced the same input variant") + } + if geminiInputVariant("run-a", 7) == geminiInputVariant("run-a", 8) { + t.Fatal("different logical indexes produced the same input variant") + } +} + +func TestAcceptanceIdempotencyKeyIncludesExecutionAndLogicalIndex(t *testing.T) { + request := httptest.NewRequest(http.MethodPost, "https://gateway.example/v1", nil) + opts := options{ + apiKeys: []string{"key-1"}, runID: "run-1", runToken: "token-1", executionID: "p24-2-gemini-baseline", + } + opts.setHeaders(request, 17, false) + if got := request.Header.Get("Idempotency-Key"); got != "acceptance-run-1-p24-2-gemini-baseline-17" { + t.Fatalf("idempotency key=%q", got) + } + poll := httptest.NewRequest(http.MethodGet, "https://gateway.example/result", nil) + opts.setHeaders(poll, 17, false) + if got := poll.Header.Get("Idempotency-Key"); got != "" { + t.Fatalf("GET request unexpectedly has idempotency key %q", got) + } +} + func TestValidateVideoAssetDownloadsFinalMedia(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { if request.Header.Get("Authorization") != "" { diff --git a/docs/operations/production-acceptance.md b/docs/operations/production-acceptance.md index bc0221c..5c029cb 100644 --- a/docs/operations/production-acceptance.md +++ b/docs/operations/production-acceptance.md @@ -8,7 +8,8 @@ 验收脚本是显式生产写操作，会暂停新正式任务、调整 Worker 容量并在恢复场景删除一个 Worker Pod。它不会自动发布新版本；只能针对已经发布、完整 SHA 和 digest 固定的 manifest -执行。进入线上前必须先通过[本地三节点同构验收](local-isomorphic-acceptance.md)。 +执行。默认先通过[本地三节点同构验收](local-isomorphic-acceptance.md)；用户明确授权在线验收时， +可以记录审计豁免并跳过本地阶段，但不能把跳过记录成通过。 ## 隔离与流量门禁 @@ -61,10 +62,12 @@ WebP、合法 4K 图片和越过 Seedance 2.0 官方输入边界的 `6144x2160` 收集器。它校验参考图数量、角色、格式、尺寸与内容哈希，并统计重复上游提交和重复回调。 默认通过 `AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS=32` 幂等准备 32 个仅属于验收用户组的 -身份、API Key 和独立钱包，并按请求序号轮询使用。这样可模拟真实多租户并发，避免把 1000 +身份、API Key 和独立钱包，并按全局请求序号轮询使用。每次负载执行都有独立 execution ID， +它同时进入幂等键和输入图片哈希，因此三次重复验收既不会重放旧任务，也不会命中前一轮的 +输入资产缓存。这样可模拟真实多租户并发，避免把 1000 个任务全部压在同一钱包行锁上；每个任务仍执行真实预留、结算和重复扣费检查。该值可在 `1..128` 内通过环境变量调整，Run 只保存 Key/User ID 对，Key Secret 仅通过 stdin 传入 -集群内压测进程，不进入 Pod spec、数据库 Run 配置、报告或日志。 +双站压测进程，不进入 Pod spec、数据库 Run 配置、报告或日志。 ## Worker 配置 @@ -96,7 +99,7 @@ AI_GATEWAY_WORKER_REPLICAS_HONGKONG 重新编码，避免执行槽升档时把大图峰值内存同比放大。2 GiB Worker 的起始值为 `2`；它不占用 视频提交后的上游等待槽，只有实测单 Pod RSS 和节点预算允许时才单独升档。 -API Pod 使用独立发布配置 `AI_GATEWAY_API_DATABASE_MAX_CONNS`，生产同构验收默认使用 `64`； +API Pod 使用独立发布配置 `AI_GATEWAY_API_DATABASE_MAX_CONNS`，生产同构验收默认使用 `31`； 也可通过 `AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS` 覆盖本次验收值，无需修改 Go 代码。 所有 API/Worker Pool 的 `AI_GATEWAY_DATABASE_MIN_IDLE_CONNS` 固定为 `4`，仅预热必要连接， `AI_GATEWAY_DATABASE_MAX_CONN_IDLE_SECONDS` 为 `30`，突发扩出的空闲连接会及时回落， @@ -135,6 +138,12 @@ PNG 并通过验收 Key 上传到 Gateway。生产快照默认在已部署的 AP `AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME=ai.51easyai.com`。压测器会继续校验正式 证书，并同时使用该域名作为 HTTP Host；这不是跳过 TLS 校验。 +线上高负载不再由操作者本机单进程跨公网发送。脚本构建同一源码的静态 linux/amd64 +压测器，校验 SHA-256 后以 `0600` 临时环境分别运行在宁波、香港宿主机的 K3s 之外；两个 +进程同时只访问本站 TLS 入口，并按奇偶全局序号分担请求。这样避免本机上行带宽成为容量 +瓶颈，同时仍覆盖正式 TLS/Nginx 入口。站点报告按请求数相加、耗时取较慢站点、p50/p95/p99 +取两站较差值进行保守合并。进程、私有环境和远端部分报告在 Run 退出时按精确路径清理。 + ```bash scripts/cluster/run-production-acceptance.sh \ --execute dist/releases/<完整SHA>.json \ diff --git a/scripts/acceptance/run-local-acceptance.sh b/scripts/acceptance/run-local-acceptance.sh index 32dafd2..26a9961 100755 --- a/scripts/acceptance/run-local-acceptance.sh +++ b/scripts/acceptance/run-local-acceptance.sh @@ -161,7 +161,8 @@ run_load() { AI_GATEWAY_ACCEPTANCE_RUN_TOKEN="$acceptance_run_token" \ AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL="$acceptance_gemini_model" \ AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL="$acceptance_video_model" \ - "$load_binary" -profile "$profile" -report "$report_path" "$@" \ + "$load_binary" -profile "$profile" -report "$report_path" \ + -execution-id "$(basename "$report_path" .json)" "$@" \ >"$stdout_temporary" & load_pid=$! active_load_pid=$load_pid diff --git a/scripts/cluster/run-production-acceptance.sh b/scripts/cluster/run-production-acceptance.sh index 72a22af..80b5f5e 100755 --- a/scripts/cluster/run-production-acceptance.sh +++ b/scripts/cluster/run-production-acceptance.sh @@ -74,7 +74,7 @@ if [[ $skip_local_acceptance != true ]]; then fi load_cluster_env -require_commands curl git go jq node openssl sed +require_commands curl git go jq node openssl sed shasum : "${AI_GATEWAY_ACCEPTANCE_ADMIN_TOKEN:=}" : "${AI_GATEWAY_ACCEPTANCE_API_KEY:=}" @@ -225,6 +225,7 @@ acceptance_group_id= acceptance_participants_json='[]' AI_GATEWAY_ACCEPTANCE_API_KEYS=$AI_GATEWAY_ACCEPTANCE_API_KEY acceptance_load_binary=$temporary_root/easyai-ai-gateway-acceptance-load +acceptance_load_linux_binary=$temporary_root/easyai-ai-gateway-acceptance-load-linux-amd64 acceptance_snapshot_binary=$temporary_root/easyai-ai-gateway-acceptance-snapshot current_production_snapshot=$temporary_root/current-production-snapshot.json local_snapshot_config_hash= @@ -235,6 +236,7 @@ if [[ $skip_local_acceptance != true ]]; then fi active_load_pid= active_pressure_pid= +remote_load_drivers_installed=false capacity_profiles_payload= image_stable_throughput= image_admitted_throughput= @@ -254,6 +256,9 @@ cleanup() { kill "$active_pressure_pid" >/dev/null 2>&1 || true wait "$active_pressure_pid" >/dev/null 2>&1 || true fi + if [[ $remote_load_drivers_installed == true ]]; then + cleanup_remote_load_drivers >/dev/null 2>&1 || true + fi if (( status != 0 )) && [[ -n $run_id && $failure_recorded != true ]]; then set +e apply_capacity_profile "$stable_profile" >/dev/null 2>&1 @@ -307,6 +312,8 @@ bootstrap_acceptance_admin_token cd "$cluster_root/apps/api" env -u AI_GATEWAY_TEST_DATABASE_URL go build \ -trimpath -o "$acceptance_load_binary" ./cmd/acceptance-load + env -u AI_GATEWAY_TEST_DATABASE_URL CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build \ + -trimpath -o "$acceptance_load_linux_binary" ./cmd/acceptance-load env -u AI_GATEWAY_TEST_DATABASE_URL go build \ -trimpath -o "$acceptance_snapshot_binary" ./cmd/acceptance-snapshot ) @@ -1750,24 +1757,14 @@ run_mixed_load_profile() { local video_rate=$4 local report_path=$5 local timeout_value=$6 - local load_stdout=$temporary_root/load-$profile-stdout.json - AI_GATEWAY_ACCEPTANCE_GATEWAYS="$AI_GATEWAY_ACCEPTANCE_GATEWAYS" \ - AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME="$AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME" \ - AI_GATEWAY_ACCEPTANCE_EMULATOR_URL='http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090' \ - AI_GATEWAY_ACCEPTANCE_API_KEYS="$AI_GATEWAY_ACCEPTANCE_API_KEYS" \ - AI_GATEWAY_ACCEPTANCE_RUN_ID="$run_id" \ - AI_GATEWAY_ACCEPTANCE_RUN_TOKEN="$run_token" \ - AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL="$AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL" \ - AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL="$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL" \ - AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS="$AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS" \ - "$acceptance_load_binary" \ - -profile "$profile" \ - -duration "$duration" \ - -image-rate "$image_rate" \ - -video-rate "$video_rate" \ - -timeout "$timeout_value" \ - -report "$report_path" >"$load_stdout" - jq -e '.secretSafe == true and .passed == true' "$report_path" >/dev/null + local site_image_rate site_video_rate + site_image_rate=$(awk -v value="$image_rate" 'BEGIN {printf "%.9f", value/2}') + site_video_rate=$(awk -v value="$video_rate" 'BEGIN {printf "%.9f", value/2}') + run_distributed_load_profile "$profile" "$report_path" \ + -duration "$duration" \ + -image-rate "$site_image_rate" \ + -video-rate "$site_video_rate" \ + -timeout "$timeout_value" } run_mixed_soak_and_overload() { @@ -1904,10 +1901,196 @@ activate_certified_autoscaling() { verify_runtime_gates } +remote_load_binary_path() { + [[ $run_id =~ ^[0-9a-f-]{36}$ ]] + printf '/root/easyai-acceptance-load-%s' "$run_id" +} + +remote_load_env_path() { + local site=$1 + [[ $site == ningbo || $site == hongkong ]] + [[ $run_id =~ ^[0-9a-f-]{36}$ ]] + printf '/root/easyai-acceptance-load-%s-%s.env' "$run_id" "$site" +} + +cleanup_remote_load_drivers() { + [[ $run_id =~ ^[0-9a-f-]{36}$ ]] || return 0 + local binary_path ningbo_env hongkong_env + binary_path=$(remote_load_binary_path) + ningbo_env=$(remote_load_env_path ningbo) + hongkong_env=$(remote_load_env_path hongkong) + cluster_ssh "$CLUSTER_NINGBO_HOST" \ + "pkill -TERM -f '^$binary_path( |$)' >/dev/null 2>&1 || true; [[ ! -e '$binary_path' ]] || unlink '$binary_path'; [[ ! -e '$ningbo_env' ]] || unlink '$ningbo_env'" || true + cluster_ssh "$CLUSTER_HONGKONG_HOST" \ + "pkill -TERM -f '^$binary_path( |$)' >/dev/null 2>&1 || true; [[ ! -e '$binary_path' ]] || unlink '$binary_path'; [[ ! -e '$hongkong_env' ]] || unlink '$hongkong_env'" || true + remote_load_drivers_installed=false +} + +write_remote_load_env() { + local output=$1 + local gateway=$2 + install -m 0600 /dev/null "$output" + { + printf 'AI_GATEWAY_ACCEPTANCE_GATEWAYS=%q\n' "$gateway" + printf 'AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME=%q\n' "$AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME" + printf 'AI_GATEWAY_ACCEPTANCE_EMULATOR_URL=%q\n' 'http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090' + printf 'AI_GATEWAY_ACCEPTANCE_API_KEYS=%q\n' "$AI_GATEWAY_ACCEPTANCE_API_KEYS" + printf 'AI_GATEWAY_ACCEPTANCE_RUN_ID=%q\n' "$run_id" + printf 'AI_GATEWAY_ACCEPTANCE_RUN_TOKEN=%q\n' "$run_token" + printf 'AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL=%q\n' "$AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL" + printf 'AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL=%q\n' "$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL" + printf 'AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS=%q\n' "$AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS" + } >>"$output" +} + +install_remote_load_drivers() { + [[ $remote_load_drivers_installed == false ]] || return 0 + local binary_path local_digest remote_digest site host gateway env_file remote_env + binary_path=$(remote_load_binary_path) + local_digest=$(shasum -a 256 "$acceptance_load_linux_binary" | awk '{print $1}') + local -a sites=(ningbo hongkong) + local -a hosts=("$CLUSTER_NINGBO_HOST" "$CLUSTER_HONGKONG_HOST") + local -a gateways=("https://${CLUSTER_NINGBO_HOST#root@}" "https://${CLUSTER_HONGKONG_HOST#root@}") + for site_index in 0 1; do + site=${sites[$site_index]} + host=${hosts[$site_index]} + gateway=${gateways[$site_index]} + env_file=$temporary_root/remote-load-$site.env + remote_env=$(remote_load_env_path "$site") + write_remote_load_env "$env_file" "$gateway" + cluster_scp "$acceptance_load_linux_binary" "$host:$binary_path" >/dev/null + cluster_scp "$env_file" "$host:$remote_env" >/dev/null + cluster_ssh "$host" "chmod 0755 '$binary_path'; chmod 0600 '$remote_env'; sha256sum '$binary_path'" >"$temporary_root/remote-load-$site.sha256" + remote_digest=$(awk '{print $1}' "$temporary_root/remote-load-$site.sha256") + [[ $remote_digest == "$local_digest" ]] + done + remote_load_drivers_installed=true + echo 'acceptance_load_drivers=PASS sites=2 placement=k3s_external_host_processes digest_verified=true' +} + +run_remote_load_site() { + local site=$1 + local host=$2 + local profile=$3 + local artifact=$4 + local shard_index=$5 + shift 5 + [[ $site == ningbo || $site == hongkong ]] + [[ $profile =~ ^[a-z-]+$ && $artifact =~ ^[a-z0-9-]+\.json$ ]] + local binary_path remote_env remote_report local_report status=0 + binary_path=$(remote_load_binary_path) + remote_env=$(remote_load_env_path "$site") + remote_report=/root/easyai-acceptance-load-"$run_id"-"$artifact" + local_report=$temporary_root/"$site"-"$artifact" + cluster_ssh "$host" bash -s -- \ + "$binary_path" "$remote_env" "$profile" "$remote_report" "$shard_index" 2 "${artifact%.json}" "$@" \ + >"$temporary_root/$site-$artifact.stdout" <<'REMOTE' || status=$? +set -euo pipefail +binary_path=$1 +env_file=$2 +profile=$3 +report=$4 +shard_index=$5 +shard_count=$6 +execution_id=$7 +shift 7 +set -a +# shellcheck source=/dev/null +source "$env_file" +set +a +"$binary_path" -profile "$profile" -report "$report" \ + -shard-index "$shard_index" -shard-count "$shard_count" \ + -execution-id "$execution_id" "$@" +REMOTE + cluster_scp "$host:$remote_report" "$local_report" >/dev/null || return 1 + cluster_ssh "$host" "[[ ! -e '$remote_report' ]] || unlink '$remote_report'" >/dev/null || true + jq -e --arg runId "$run_id" --arg profile "$profile" \ + '.schemaVersion == "acceptance-load-report/v1" and .runId == $runId and .profile == $profile and .secretSafe == true' \ + "$local_report" >/dev/null || return 1 + return "$status" +} + +merge_remote_load_reports() { + local profile=$1 + local ningbo_report=$2 + local hongkong_report=$3 + local output=$4 + jq -s --arg profile "$profile" ' + def sum_field($name): [.[].phases[0][$name] // 0] | add; + def max_field($name): [.[].phases[0][$name] // 0] | max; + . as $reports + | { + schemaVersion:"acceptance-load-report/v1", + runId:$reports[0].runId, + profile:$profile, + startedAt:([$reports[].startedAt] | min), + finishedAt:([$reports[].finishedAt] | max), + passed:([$reports[].passed] | all), + phases:[{ + name:$profile, + requests:sum_field("requests"), + completed:sum_field("completed"), + failed:sum_field("failed"), + durationMs:max_field("durationMs"), + submissionDurationMs:max_field("submissionDurationMs"), + p50Ms:max_field("p50Ms"), + p95Ms:max_field("p95Ms"), + p99Ms:max_field("p99Ms"), + decodedOutputBytes:sum_field("decodedOutputBytes"), + outputSha256:([$reports[].phases[0].outputSha256 // ""] | map(select(length>0)) | first // ""), + uniqueTaskIds:sum_field("uniqueTaskIds"), + uniqueImageCombinations:max_field("uniqueImageCombinations"), + forcedConversionTasks:sum_field("forcedConversionTasks"), + throttled:sum_field("throttled"), + unexpected5xx:sum_field("unexpected5xx"), + offeredRatePerSecond:sum_field("offeredRatePerSecond") + }], + failure:([$reports[] | select(.passed == false) | .failure] | first // null), + failureOperation:([$reports[] | select(.passed == false) | .failureOperation] | first // null), + aggregation:{ + sites:["ningbo","hongkong"], + percentileMethod:"conservative_max_of_site_percentiles", + durationMethod:"max_site_duration", + placement:"k3s_external_host_processes" + }, + secretSafe:true + } + | with_entries(select(.value != null)) + ' "$ningbo_report" "$hongkong_report" >"$output" + chmod 0600 "$output" +} + +run_distributed_load_profile() { + local profile=$1 + local report_path=$2 + shift 2 + install_remote_load_drivers + local artifact base ningbo_report hongkong_report ningbo_pid hongkong_pid + local ningbo_status=0 hongkong_status=0 + base=$(basename "$report_path" .json) + artifact=$base.json + run_remote_load_site ningbo "$CLUSTER_NINGBO_HOST" "$profile" "$artifact" 0 "$@" & + ningbo_pid=$! + run_remote_load_site hongkong "$CLUSTER_HONGKONG_HOST" "$profile" "$artifact" 1 "$@" & + hongkong_pid=$! + wait "$ningbo_pid" || ningbo_status=$? + wait "$hongkong_pid" || hongkong_status=$? + ningbo_report=$temporary_root/ningbo-$artifact + hongkong_report=$temporary_root/hongkong-$artifact + [[ -f $ningbo_report && -f $hongkong_report ]] || return 1 + merge_remote_load_reports "$profile" "$ningbo_report" "$hongkong_report" "$report_path" + (( ningbo_status == 0 && hongkong_status == 0 )) || return 1 + jq -e '.secretSafe == true and .passed == true' "$report_path" >/dev/null +} + run_load_profile() { local profile=$1 local report_path=$2 [[ $profile =~ ^[a-z-]+$ ]] || return 1 + if [[ $profile != real-canary ]]; then + run_distributed_load_profile "$profile" "$report_path" + return + fi local load_stdout=$temporary_root/load-$profile-stdout.json AI_GATEWAY_ACCEPTANCE_GATEWAYS="$AI_GATEWAY_ACCEPTANCE_GATEWAYS" \ AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME="$AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME" \ @@ -1919,6 +2102,7 @@ run_load_profile() { AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL="$AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL" \ AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS="$AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS" \ "$acceptance_load_binary" -profile "$profile" -report "$report_path" \ + -execution-id "$(basename "$report_path" .json)" \ >"$load_stdout" jq -e '.secretSafe == true and .passed == true' "$report_path" >/dev/null } @@ -2617,6 +2801,7 @@ finish_and_report() { local gemini_invalid video_invalid reference_count_3 reference_count_6 reference_count_9 local reference_role_count first_frame_role_count last_frame_role_count local unique_image_hashes unique_video_tasks + local unique_gemini_input_hashes direct_gemini_input_assets local duplicate_remote_tasks duplicate_terminal_events duplicate_settlements local duplicate_wallet_transactions duplicate_callback_rows pending_settlements pending_callbacks local missing_billings @@ -2625,6 +2810,19 @@ finish_and_report() { gemini_task_count=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND kind='images.edits' AND run_mode='acceptance';") forced_tasks=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND request::text LIKE '%acceptance-force-conversion%';") compact_payloads=$(database_query "SELECT count(*) FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid AND kind='images.edits' AND (pg_column_size(request) >= 1048576 OR pg_column_size(result) >= 1048576);") + unique_gemini_input_hashes=$(database_query " +SELECT count(DISTINCT request #>> '{image,assetRef,sha256}') +FROM gateway_tasks +WHERE acceptance_run_id='$run_id'::uuid + AND kind='images.edits' + AND run_mode='acceptance';") + direct_gemini_input_assets=$(database_query " +SELECT count(*) +FROM gateway_tasks +WHERE acceptance_run_id='$run_id'::uuid + AND kind='images.edits' + AND run_mode='acceptance' + AND request #>> '{image,assetRef,storageProvider}' = 'aliyun_oss_direct';") submissions=$(jq -r '.videoSubmissions' "$emulator_report") duplicates=$(jq -r '.duplicates' "$callback_report") callbacks=$(jq -r '.deliveries' "$callback_report") @@ -2653,6 +2851,10 @@ finish_and_report() { $missing_idempotency == 0 && $compact_payloads == 0 ]] || fail_acceptance_gate idempotency_and_materialization \ 'duplicate submission, missing idempotency key, or inline database media detected' + [[ $unique_gemini_input_hashes == "$gemini_task_count" && + $direct_gemini_input_assets == "$gemini_task_count" ]] || + fail_acceptance_gate gemini_input_materialization \ + 'Gemini inputs were reused across executions or bypassed direct OSS materialization' [[ $gemini_invalid == 0 && $video_invalid == 0 ]] || fail_acceptance_gate provider_protocol_validity 'Gemini or Volces protocol validation failed' [[ $reference_count_3 -gt 0 && $reference_count_6 -gt 0 && $reference_count_9 -gt 0 ]] || @@ -2796,6 +2998,8 @@ WHERE task.acceptance_run_id='$run_id'::uuid --argjson lastFrameRoleCount "$last_frame_role_count" \ --argjson uniqueImageHashes "$unique_image_hashes" \ --argjson uniqueVideoTasks "$unique_video_tasks" \ + --argjson uniqueGeminiInputHashes "$unique_gemini_input_hashes" \ + --argjson directGeminiInputAssets "$direct_gemini_input_assets" \ --argjson duplicateRemoteTasks "$duplicate_remote_tasks" \ --argjson duplicateTerminalEvents "$duplicate_terminal_events" \ --argjson duplicateSettlements "$duplicate_settlements" \ @@ -2838,7 +3042,9 @@ WHERE task.acceptance_run_id='$run_id'::uuid last_frame:$lastFrameRoleCount }, uniqueImageHashes:$uniqueImageHashes, - uniqueVideoTasks:$uniqueVideoTasks + uniqueVideoTasks:$uniqueVideoTasks, + uniqueGeminiInputHashes:$uniqueGeminiInputHashes, + directGeminiInputAssets:$directGeminiInputAssets }, databaseConsistency:{ duplicateRemoteTasks:$duplicateRemoteTasks,