From edcbbd6a878315dabcf92c1ef7c995e70d756430 Mon Sep 17 00:00:00 2001 From: wangbo Date: Sat, 1 Aug 2026 18:23:00 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20=E6=8B=86=E5=88=86=E5=87=86?= =?UTF-8?q?=E5=85=A5=E4=BA=8B=E5=8A=A1=E5=B9=B6=E6=94=B6=E6=95=9B=E9=AA=8C?= =?UTF-8?q?=E6=94=B6=E5=A4=B1=E8=B4=A5=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将批量准入改为每任务独立事务，避免 P24 压测时 48 条任务共享长事务造成 transactionid 锁潮与 Worker 槽位空转。失败验收在保持 validation 的同时清理未提交任务，并记录精确压力门禁原因。统一生产 0+2 Worker 拓扑、运行时 ConfigMap 与发布配置，补充镜像预检诊断和回归测试。\n\n验证：Go 全量测试、go vet、真实 PostgreSQL 48 任务集成测试、迁移安全检查、发布脚本测试、bash -n、ShellCheck、Compose 配置均通过。 --- apps/api/internal/runner/admission.go | 7 +- apps/api/internal/store/acceptance.go | 24 ++++- .../store/acceptance_integration_test.go | 14 +++ apps/api/internal/store/admission_queue.go | 83 ++++++------------ .../store/admission_queue_integration_test.go | 10 ++- .../easyai-ai-gateway-cluster-release | 87 +++++++++++++++---- ...ai-ai-gateway-cluster-release.conf.example | 10 +-- .../production/application-config.yaml | 10 +-- deploy/kubernetes/production/application.yaml | 4 +- scripts/cluster/run-production-acceptance.sh | 41 +++++++-- .../production-acceptance-script-test.sh | 4 +- 11 files changed, 194 insertions(+), 100 deletions(-) diff --git a/apps/api/internal/runner/admission.go b/apps/api/internal/runner/admission.go index 35d6868..6954f9d 100644 --- a/apps/api/internal/runner/admission.go +++ b/apps/api/internal/runner/admission.go @@ -33,10 +33,9 @@ const ( asyncWorkerCapacityScopeKey = "global" asyncWorkerQueueLimit = 10000 asyncWorkerMaxWaitSeconds = 24 * 60 * 60 - // One transaction must be able to fill the certified 2 x P32 global - // capacity. Smaller windows let multiple Worker dispatchers repeatedly race - // on the same FIFO head and can turn a batch size into an accidental cluster - // concurrency ceiling. + // The dispatcher scans enough FIFO entries to fill the certified 2 x P32 + // capacity. Store admission commits each item independently so this scan + // size is never also a multi-task transaction size. asyncAdmissionBatchMax = 64 acceptanceQueueLimit = 10000 acceptanceQueueMaxWait = 15 * 60 diff --git a/apps/api/internal/store/acceptance.go b/apps/api/internal/store/acceptance.go index 3345a11..2f15291 100644 --- a/apps/api/internal/store/acceptance.go +++ b/apps/api/internal/store/acceptance.go @@ -258,7 +258,12 @@ func (s *Store) FinishAcceptanceRun(ctx context.Context, input FinishAcceptanceR status = "passed" } report, _ := json.Marshal(sanitizeAcceptanceMetadata(sanitizeJSONForStorage(input.Report))) - return scanAcceptanceRun(s.pool.QueryRow(ctx, ` + tx, err := s.pool.Begin(ctx) + if err != nil { + return AcceptanceRun{}, err + } + defer rollbackTransaction(tx) + run, err := scanAcceptanceRun(tx.QueryRow(ctx, ` UPDATE gateway_acceptance_runs SET status = $2, report = $3::jsonb, failure_reason = NULLIF($4, ''), finished_at = now(), updated_at = now() @@ -287,6 +292,23 @@ RETURNING `+acceptanceRunColumns, strings.TrimSpace(input.RunID), status, report, strings.TrimSpace(input.FailureReason), SystemSettingGatewayTrafficMode, )) + if err != nil { + return AcceptanceRun{}, err + } + cancelled := int64(0) + if !input.Passed { + cancelled, err = cancelSafeAcceptanceTasksTx(ctx, tx, run.ID) + if err != nil { + return AcceptanceRun{}, err + } + } + if err := tx.Commit(ctx); err != nil { + return AcceptanceRun{}, err + } + if cancelled > 0 { + s.notifyTaskAdmissionBestEffort(ctx, "*") + } + return run, nil } func (s *Store) RetryAcceptanceRun(ctx context.Context, runID string) (AcceptanceRun, error) { diff --git a/apps/api/internal/store/acceptance_integration_test.go b/apps/api/internal/store/acceptance_integration_test.go index bdaf6c8..01021b7 100644 --- a/apps/api/internal/store/acceptance_integration_test.go +++ b/apps/api/internal/store/acceptance_integration_test.go @@ -252,6 +252,20 @@ WHERE id = $1::uuid`, submittingAttemptID); err != nil { }); err != nil || failed.Status != "failed" { t.Fatalf("downgrade unpromoted acceptance run after external gate failure=%+v err=%v", failed, err) } + var failedRunCancelled int + if err := db.pool.QueryRow(ctx, ` +SELECT count(*) +FROM gateway_tasks +WHERE id = ANY($1::uuid[]) + AND status = 'cancelled' + AND error_code = 'acceptance_run_aborted'`, + []string{queuedTask.ID, runningTask.ID, notSubmittedTask.ID}, + ).Scan(&failedRunCancelled); err != nil { + t.Fatalf("read failed acceptance cleanup states: %v", err) + } + if failedRunCancelled != 3 { + t.Fatalf("failed acceptance cleanup cancelled=%d, want 3", failedRunCancelled) + } if retried, err := db.RetryAcceptanceRun(ctx, failedRun.ID); err != nil || retried.Status != "running" { t.Fatalf("retry acceptance run=%+v err=%v", retried, err) } diff --git a/apps/api/internal/store/admission_queue.go b/apps/api/internal/store/admission_queue.go index b586bda..6d3528d 100644 --- a/apps/api/internal/store/admission_queue.go +++ b/apps/api/internal/store/admission_queue.go @@ -455,10 +455,12 @@ WHERE id = $1::uuid`, input.TaskID).Scan(&taskActive); err != nil { }, nil } -// TryTaskAdmissionBatchWithAdmittedHook fills a FIFO capacity window in one -// transaction. The hook inserts every unique River job in that same -// transaction, preserving the admission/job crash boundary while avoiding one -// synchronous replication round trip per task. +// TryTaskAdmissionBatchWithAdmittedHook fills a FIFO capacity window with one +// transaction per task. Each transaction still atomically reserves admission +// and inserts its unique River job, but a large global capacity window no +// longer holds every task row and scope lock until the entire batch commits. +// This is deliberately a batch at the dispatcher boundary, not a database +// transaction boundary. func (s *Store) TryTaskAdmissionBatchWithAdmittedHook( ctx context.Context, inputs []TaskAdmissionInput, @@ -467,68 +469,37 @@ func (s *Store) TryTaskAdmissionBatchWithAdmittedHook( if len(inputs) == 0 { return nil, nil } - lockKeys := make([]string, 0, len(inputs)*3) for _, input := range inputs { if err := validateTaskAdmissionInput(input); err != nil { return nil, err } - lockKeys = append(lockKeys, admissionOperationLockKeys(input)...) } - lockKeys = normalizedAdmissionLockKeys(lockKeys) - return retryAdmissionOperation(ctx, lockKeys, func() ([]TaskAdmissionBatchOutcome, error) { - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, err - } - defer rollbackTransaction(tx) - // A batch touches one shared capacity scope and multiple task keys. Lock - // the complete union in one global order before processing any task. If - // two API processes dispatch overlapping FIFO windows, neither can hold a - // scope while waiting on a task key already owned by the other batch. - for _, lockKey := range lockKeys { - if err := tryAdmissionTransactionLock(ctx, tx, lockKey); err != nil { - return nil, err + outcomes := make([]TaskAdmissionBatchOutcome, 0, len(inputs)) + for _, input := range inputs { + hook := func(tx pgx.Tx) error { + if onAdmitted == nil { + return nil } + return onAdmitted(tx, input) } - - outcomes := make([]TaskAdmissionBatchOutcome, 0, len(inputs)) - notify := false - for _, input := range inputs { - hook := func(tx pgx.Tx) error { - if onAdmitted == nil { - return nil - } - return onAdmitted(tx, input) - } - outcome, admissionErr := s.tryTaskAdmissionTx(ctx, tx, input, hook) - if errors.Is(admissionErr, ErrTaskExecutionFinished) { - outcomes = append(outcomes, TaskAdmissionBatchOutcome{ - TaskID: input.TaskID, - Err: admissionErr, - }) - continue - } - if admissionErr != nil { - return nil, admissionErr - } - outcomes = append(outcomes, TaskAdmissionBatchOutcome{ - TaskID: input.TaskID, - Result: outcome.Result, - Err: outcome.PostCommitErr, - }) - notify = notify || outcome.NotifyTaskID != "" - if outcome.PostCommitErr == nil && !outcome.Result.Admitted { - break - } + result, admissionErr := s.TryTaskAdmissionWithAdmittedHook(ctx, input, hook) + outcomes = append(outcomes, TaskAdmissionBatchOutcome{ + TaskID: input.TaskID, + Result: result, + Err: admissionErr, + }) + if errors.Is(admissionErr, ErrTaskExecutionFinished) || + errors.Is(admissionErr, ErrQueueTimeout) { + continue } - if err := tx.Commit(ctx); err != nil { - return nil, err + if admissionErr != nil { + return outcomes, admissionErr } - if notify { - s.notifyTaskAdmissionBestEffort(ctx, "*") + if !result.Admitted { + break } - return outcomes, nil - }) + } + return outcomes, nil } func validateNewAdmissionCapacity(states []admissionScopeState) error { diff --git a/apps/api/internal/store/admission_queue_integration_test.go b/apps/api/internal/store/admission_queue_integration_test.go index 356c902..c0b5b6b 100644 --- a/apps/api/internal/store/admission_queue_integration_test.go +++ b/apps/api/internal/store/admission_queue_integration_test.go @@ -598,7 +598,7 @@ SELECT rollbackInputs = append(rollbackInputs, input) } batchHookFailure := errors.New("synthetic batch hook failure") - _, err = first.TryTaskAdmissionBatchWithAdmittedHook( + partialOutcomes, err := first.TryTaskAdmissionBatchWithAdmittedHook( ctx, rollbackInputs, func(tx pgx.Tx, input TaskAdmissionInput) error { @@ -615,6 +615,10 @@ WHERE id = $1::uuid`, input.TaskID) if !errors.Is(err, batchHookFailure) { t.Fatalf("batch hook failure=%v, want synthetic failure", err) } + if len(partialOutcomes) != 2 || !partialOutcomes[0].Result.Admitted || + !errors.Is(partialOutcomes[1].Err, batchHookFailure) { + t.Fatalf("per-task batch outcomes=%+v, want first committed and second failed", partialOutcomes) + } var rollbackWaiting, rollbackLeases, rollbackRiverJobs int if err := first.pool.QueryRow(ctx, ` SELECT @@ -631,9 +635,9 @@ SELECT ).Scan(&rollbackWaiting, &rollbackLeases, &rollbackRiverJobs); err != nil { t.Fatalf("read rolled back batch admission: %v", err) } - if rollbackWaiting != 2 || rollbackLeases != 0 || rollbackRiverJobs != 0 { + if rollbackWaiting != 1 || rollbackLeases != 1 || rollbackRiverJobs != 1 { t.Fatalf( - "rolled back batch waiting=%d leases=%d River jobs=%d, want 2/0/0", + "per-task rollback waiting=%d leases=%d River jobs=%d, want 1/1/1", rollbackWaiting, rollbackLeases, rollbackRiverJobs, diff --git a/deploy/kubernetes/easyai-ai-gateway-cluster-release b/deploy/kubernetes/easyai-ai-gateway-cluster-release index 302d494..22e82db 100755 --- a/deploy/kubernetes/easyai-ai-gateway-cluster-release +++ b/deploy/kubernetes/easyai-ai-gateway-cluster-release @@ -15,8 +15,8 @@ source "$config_file" : "${PUBLIC_BASE_URL:=https://ai.51easyai.com}" : "${K3S_BIN:=/usr/local/bin/k3s}" : "${DESIRED_STATE_DIR:=/usr/local/share/easyai-ai-gateway-release/production}" -: "${AI_GATEWAY_WORKER_REPLICAS_NINGBO:=1}" -: "${AI_GATEWAY_WORKER_REPLICAS_HONGKONG:=1}" +: "${AI_GATEWAY_WORKER_REPLICAS_NINGBO:=0}" +: "${AI_GATEWAY_WORKER_REPLICAS_HONGKONG:=2}" : "${AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT:=24}" : "${AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT:=48}" : "${AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT:=$AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT}" @@ -32,10 +32,10 @@ source "$config_file" : "${AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY:=24}" : "${AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY:=24}" : "${AI_GATEWAY_WORKER_AUTOSCALING_ENABLED:=false}" -: "${AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO:=1}" +: "${AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO:=0}" : "${AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG:=1}" -: "${AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO:=1}" -: "${AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG:=1}" +: "${AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO:=0}" +: "${AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG:=2}" : "${AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA:=$((AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT * 2))}" : "${AI_GATEWAY_WORKER_SCALE_UP_WINDOW_SECONDS:=20}" : "${AI_GATEWAY_WORKER_SCALE_DOWN_STABILIZATION_SECONDS:=600}" @@ -199,8 +199,55 @@ capacity_config_hash() { } record_capacity_config() { - local config_hash - config_hash=$(capacity_config_hash) + local config_hash + local runtime_patch + runtime_patch=$(jq -nc \ + --arg replicasNingbo "$AI_GATEWAY_WORKER_REPLICAS_NINGBO" \ + --arg replicasHongkong "$AI_GATEWAY_WORKER_REPLICAS_HONGKONG" \ + --arg minNingbo "$AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO" \ + --arg minHongkong "$AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG" \ + --arg maxNingbo "$AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO" \ + --arg maxHongkong "$AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG" \ + --arg autoscaling "$AI_GATEWAY_WORKER_AUTOSCALING_ENABLED" \ + --arg instanceLimit "$AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT" \ + --arg globalLimit "$AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT" \ + --arg workerPool "$AI_GATEWAY_DATABASE_MAX_CONNS" \ + --arg materialization "$AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY" \ + --arg mediaRequest "$AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY" \ + --arg targetOutstanding "$AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA" \ + --arg scaleUp "$AI_GATEWAY_WORKER_SCALE_UP_WINDOW_SECONDS" \ + --arg scaleDown "$AI_GATEWAY_WORKER_SCALE_DOWN_STABILIZATION_SECONDS" \ + --arg drainTimeout "$AI_GATEWAY_WORKER_DRAIN_TIMEOUT_SECONDS" \ + --arg memoryTarget "$AI_GATEWAY_NODE_MEMORY_TARGET_PERCENT" \ + --arg memoryHard "$AI_GATEWAY_NODE_MEMORY_HARD_PERCENT" \ + --arg cpuTarget "$AI_GATEWAY_NODE_CPU_TARGET_PERCENT" \ + --arg postgresBudget "$AI_GATEWAY_POSTGRES_CONNECTION_BUDGET" \ + '{data:{ + AI_GATEWAY_WORKER_REPLICAS_NINGBO:$replicasNingbo, + AI_GATEWAY_WORKER_REPLICAS_HONGKONG:$replicasHongkong, + AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO:$minNingbo, + AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG:$minHongkong, + AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO:$maxNingbo, + AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG:$maxHongkong, + AI_GATEWAY_WORKER_AUTOSCALING_ENABLED:$autoscaling, + AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT:$instanceLimit, + AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT:$globalLimit, + AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT:$globalLimit, + AI_GATEWAY_WORKER_DATABASE_MAX_CONNS:$workerPool, + AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY:$materialization, + AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY:$mediaRequest, + AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA:$targetOutstanding, + AI_GATEWAY_WORKER_SCALE_UP_WINDOW_SECONDS:$scaleUp, + AI_GATEWAY_WORKER_SCALE_DOWN_STABILIZATION_SECONDS:$scaleDown, + AI_GATEWAY_WORKER_DRAIN_TIMEOUT_SECONDS:$drainTimeout, + AI_GATEWAY_NODE_MEMORY_TARGET_PERCENT:$memoryTarget, + AI_GATEWAY_NODE_MEMORY_HARD_PERCENT:$memoryHard, + AI_GATEWAY_NODE_CPU_TARGET_PERCENT:$cpuTarget, + AI_GATEWAY_POSTGRES_CONNECTION_BUDGET:$postgresBudget + }}') + "${kubectl[@]}" patch configmap easyai-ai-gateway-config -n "$NAMESPACE" \ + --type merge -p "$runtime_patch" >/dev/null + config_hash=$(capacity_config_hash) "${kubectl[@]}" patch configmap easyai-gateway-release -n "$NAMESPACE" \ --type merge -p "{\"data\":{\"configHash\":\"$config_hash\"}}" >/dev/null "${kubectl[@]}" annotate namespace "$NAMESPACE" \ @@ -512,9 +559,18 @@ spec: emptyDir: sizeLimit: 64Mi EOF - if ! "${kubectl[@]}" wait --for=condition=Ready "pod/$pod" \ - -n "$NAMESPACE" --timeout=120s; then - "${kubectl[@]}" delete pod "$pod" -n "$NAMESPACE" \ + if ! "${kubectl[@]}" wait --for=condition=Ready "pod/$pod" \ + -n "$NAMESPACE" --timeout=300s; then + "${kubectl[@]}" get pod "$pod" -n "$NAMESPACE" -o json 2>/dev/null | jq -c ' + { + phase:(.status.phase // "unknown"), + node:(.spec.nodeName // "unscheduled"), + ready:([.status.containerStatuses[]?.ready] | all), + restarts:([.status.containerStatuses[]?.restartCount] | add // 0), + waitingReasons:[.status.containerStatuses[]?.state.waiting.reason // empty], + terminatedReasons:[.status.containerStatuses[]?.state.terminated.reason // empty] + }' >&2 || true + "${kubectl[@]}" delete pod "$pod" -n "$NAMESPACE" \ --ignore-not-found=true --wait=true >/dev/null || true echo 'capacity controller exact-image preflight failed' >&2 return 1 @@ -866,7 +922,7 @@ activate_manifest() { -n "$NAMESPACE" --replicas=0 fi - if { [[ $api_changed != true ]] || + if { [[ $api_changed != true ]] || { rollout_worker_site hongkong "$api_image" && rollout_worker_site ningbo "$api_image" && { [[ $action == rollback ]] || @@ -875,16 +931,17 @@ activate_manifest() { rollout_site ningbo "$api_image" "$web_image" "$api_changed" "$web_changed" && wait_for_url 'public readiness' "$PUBLIC_BASE_URL/api/v1/readyz" '"ok":true'; then : - else + else if [[ -n $previous_api && -n $previous_web ]]; then echo '[cluster-release] restoring exact pre-release Deployment snapshot' >&2 restore_application_deployments "$deployment_snapshot" || true fi rm -f -- "$deployment_snapshot" "$current_file" - return 1 - fi + return 1 + fi - record_current "$manifest" "$action" "$started_at" + record_capacity_config + record_current "$manifest" "$action" "$started_at" rm -f -- "$deployment_snapshot" "$current_file" echo "production_release=PASS action=$action release=$source_sha" } diff --git a/deploy/kubernetes/easyai-ai-gateway-cluster-release.conf.example b/deploy/kubernetes/easyai-ai-gateway-cluster-release.conf.example index a643053..802e082 100644 --- a/deploy/kubernetes/easyai-ai-gateway-cluster-release.conf.example +++ b/deploy/kubernetes/easyai-ai-gateway-cluster-release.conf.example @@ -6,8 +6,8 @@ DESIRED_STATE_DIR=/usr/local/share/easyai-ai-gateway-release/production # Worker 容量只需修改本发布配置并执行获授权的 `capacity`，不再修改 Go 代码。 # 副本数属于 Deployment 配置，不能由容器内部环境变量改变；发布工具读取这两个变量并写入 replicas。 -AI_GATEWAY_WORKER_REPLICAS_NINGBO=1 -AI_GATEWAY_WORKER_REPLICAS_HONGKONG=1 +AI_GATEWAY_WORKER_REPLICAS_NINGBO=0 +AI_GATEWAY_WORKER_REPLICAS_HONGKONG=2 AI_GATEWAY_ASYNC_WORKER_INSTANCE_HARD_LIMIT=24 AI_GATEWAY_ASYNC_WORKER_HARD_LIMIT=48 AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT=48 @@ -23,10 +23,10 @@ AI_GATEWAY_DATABASE_MAX_CONN_IDLE_SECONDS=30 AI_GATEWAY_MEDIA_MATERIALIZATION_CONCURRENCY=24 AI_GATEWAY_MEDIA_REQUEST_CONCURRENCY=24 AI_GATEWAY_WORKER_AUTOSCALING_ENABLED=false -AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO=1 +AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO=0 AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG=1 -AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO=1 -AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG=1 +AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO=0 +AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG=2 AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA=48 AI_GATEWAY_WORKER_SCALE_UP_WINDOW_SECONDS=20 AI_GATEWAY_WORKER_SCALE_DOWN_STABILIZATION_SECONDS=600 diff --git a/deploy/kubernetes/production/application-config.yaml b/deploy/kubernetes/production/application-config.yaml index abd4017..5a1f4a5 100644 --- a/deploy/kubernetes/production/application-config.yaml +++ b/deploy/kubernetes/production/application-config.yaml @@ -46,12 +46,12 @@ data: AI_GATEWAY_ASYNC_WORKER_GLOBAL_HARD_LIMIT: "48" AI_GATEWAY_ASYNC_WORKER_REFRESH_INTERVAL_SECONDS: "5" AI_GATEWAY_WORKER_AUTOSCALING_ENABLED: "false" - AI_GATEWAY_WORKER_REPLICAS_NINGBO: "1" - AI_GATEWAY_WORKER_REPLICAS_HONGKONG: "1" - AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO: "1" + AI_GATEWAY_WORKER_REPLICAS_NINGBO: "0" + AI_GATEWAY_WORKER_REPLICAS_HONGKONG: "2" + AI_GATEWAY_WORKER_MIN_REPLICAS_NINGBO: "0" AI_GATEWAY_WORKER_MIN_REPLICAS_HONGKONG: "1" - AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO: "1" - AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG: "1" + AI_GATEWAY_WORKER_MAX_REPLICAS_NINGBO: "0" + AI_GATEWAY_WORKER_MAX_REPLICAS_HONGKONG: "2" AI_GATEWAY_WORKER_TARGET_OUTSTANDING_PER_REPLICA: "48" AI_GATEWAY_WORKER_SCALE_UP_WINDOW_SECONDS: "20" AI_GATEWAY_WORKER_SCALE_DOWN_STABILIZATION_SECONDS: "600" diff --git a/deploy/kubernetes/production/application.yaml b/deploy/kubernetes/production/application.yaml index e479f0f..ed97c65 100644 --- a/deploy/kubernetes/production/application.yaml +++ b/deploy/kubernetes/production/application.yaml @@ -434,7 +434,7 @@ metadata: app.kubernetes.io/component: worker app.kubernetes.io/part-of: easyai-ai-gateway spec: - replicas: 1 + replicas: 0 strategy: type: RollingUpdate rollingUpdate: @@ -607,7 +607,7 @@ metadata: app.kubernetes.io/component: worker app.kubernetes.io/part-of: easyai-ai-gateway spec: - replicas: 1 + replicas: 2 strategy: type: RollingUpdate rollingUpdate: diff --git a/scripts/cluster/run-production-acceptance.sh b/scripts/cluster/run-production-acceptance.sh index f50c3a3..fa60385 100755 --- a/scripts/cluster/run-production-acceptance.sh +++ b/scripts/cluster/run-production-acceptance.sh @@ -1512,6 +1512,9 @@ mark_run_failed() { --output "$report_root/acceptance-report.partial.json" >/dev/null || true if admin_request POST "/api/admin/system/acceptance/runs/$run_id/finish" "$body" "$response"; then failure_recorded=true + if ! wait_for_terminal_acceptance_tasks_to_drain; then + echo 'failed acceptance Run did not drain all safe tasks and outbox work' >&2 + fi fi } @@ -2608,15 +2611,25 @@ run_with_pressure_monitor() { return "$load_status" } -pressure_row_passes_live_gates() { +pressure_row_live_gate_reason() { local pressure_row=$1 awk -F',' ' - $4 > 900 || $6 <= 0 || $5 * 4 >= $6 * 3 || - $7 >= 80 || $8 >= 1536 || $14 > 0 || $15 > 0 || - $16 < 1 || $16 > 2 || $28 >= 1536 {exit 1} + $4 >= 900 {print "oldest_wait_seconds=" $4 " limit=900"; exit} + $6 <= 0 {print "postgres_max_connections_unavailable"; exit} + $5 * 4 >= $6 * 3 {print "postgres_connections=" $5 " limit_75_percent_of=" $6; exit} + $7 >= 80 {print "node_memory_percent=" $7 " limit=80"; exit} + $8 >= 1536 {print "api_pod_memory_mib=" $8 " limit=1536"; exit} + $14 > 0 {print "lease_renewal_failures=" $14; exit} + $15 > 0 {print "lease_lost=" $15; exit} + $16 < 1 || $16 > 2 {print "active_instances=" $16 " expected=1..2"; exit} + $28 >= 1536 {print "worker_memory_mib=" $28 " limit=1536"; exit} ' <<<"$pressure_row" } +pressure_row_passes_live_gates() { + [[ -z $(pressure_row_live_gate_reason "$1") ]] +} + current_max_pod_memory_mib() { remote_kubectl top pods -n "$namespace" \ -l 'app.kubernetes.io/part-of=easyai-ai-gateway' --no-headers | @@ -2914,6 +2927,7 @@ WHERE status='active' sample_pressure() { local output=$1 local stop_file=$2 + local failure_file=${3:-} local database_state node_memory_percent pod_memory_mib worker_resources pool_state metrics local consecutive_failures=0 missing_fields metrics_status pressure_row echo 'timestamp,queued,running,oldest_wait_seconds,db_connections,db_max_connections,node_memory_percent,pod_memory_mib,pool_acquired_max,pool_max,pool_idle_min,empty_acquire_total,canceled_acquire_total,lease_failure_total,lease_lost_total,active_instances_min,allocated_capacity_max,critical_pool_acquired_max,critical_pool_max,critical_pool_idle_min,critical_empty_acquire_total,critical_canceled_acquire_total,river_pool_acquired_max,river_pool_max,river_pool_idle_min,river_empty_acquire_total,river_canceled_acquire_total,worker_memory_mib,worker_cpu_millicores' >"$output" @@ -3046,7 +3060,12 @@ WHERE acceptance_run_id='$run_id'::uuid;"); then "$pod_memory_mib" "$pool_state" "$worker_resources" printf '%s\n' "$pressure_row" >>"$output" if ! pressure_row_passes_live_gates "$pressure_row"; then - echo 'pressure sampling crossed a live resource or lease hard gate' >&2 + local gate_reason + gate_reason=$(pressure_row_live_gate_reason "$pressure_row") + if [[ -n $failure_file ]]; then + printf '%s\n' "$gate_reason" >"$failure_file" + fi + echo "pressure sampling crossed a live resource or lease hard gate: $gate_reason" >&2 return 1 fi sleep 5 @@ -3174,6 +3193,7 @@ run_capacity_round() { profile_slug=$(printf '%s' "$profile" | tr '[:upper:]' '[:lower:]') local prefix=$report_root/$profile_slug-$repetition local pressure_report=$prefix-pressure.csv + local pressure_failure=$prefix-pressure.failure.txt local pressure_stop=$temporary_root/pressure-stop local pressure_pid status=0 pressure_status=0 local baseline_memory_mib @@ -3185,7 +3205,8 @@ run_capacity_round() { esac baseline_memory_mib=$(current_max_pod_memory_mib) rm -f -- "$pressure_stop" - sample_pressure "$pressure_report" "$pressure_stop" & + rm -f -- "$pressure_failure" + sample_pressure "$pressure_report" "$pressure_stop" "$pressure_failure" & pressure_pid=$! active_pressure_pid=$pressure_pid run_with_pressure_monitor "$pressure_pid" \ @@ -3210,6 +3231,10 @@ run_capacity_round() { wait "$pressure_pid" || pressure_status=$? active_pressure_pid= (( status == 0 && pressure_status == 0 )) || { + if [[ -s $pressure_failure ]]; then + failure_gate_id=pressure_live_gate + failure_reason=$(<"$pressure_failure") + fi (( status != 0 )) && return "$status" return "$pressure_status" } @@ -3585,8 +3610,8 @@ for profile in P24 P28 P32; do for repetition in 1 2 3; do if ! run_capacity_round "$profile" "$repetition"; then profile_passed=false - failure_gate_id=capacity_round - failure_reason="$profile repetition $repetition failed" + [[ -n $failure_gate_id ]] || failure_gate_id=capacity_round + [[ -n $failure_reason ]] || failure_reason="$profile repetition $repetition failed" break fi done diff --git a/tests/release/production-acceptance-script-test.sh b/tests/release/production-acceptance-script-test.sh index 1fd4b39..dd02bbb 100755 --- a/tests/release/production-acceptance-script-test.sh +++ b/tests/release/production-acceptance-script-test.sh @@ -108,8 +108,10 @@ grep -Fq '# target_host root@hongkong-worker-2.invalid' <<<"$metrics" [[ $(metric_max test_metric <<<"$metrics") == 1 ]] [[ $(metric_sum test_metric <<<"$metrics") == 2 ]] -passing_pressure_row='2026-08-01T00:00:00Z,0,0,900,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' +passing_pressure_row='2026-08-01T00:00:00Z,0,0,899,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' pressure_row_passes_live_gates "$passing_pressure_row" +[[ $(pressure_row_live_gate_reason '2026-08-01T00:00:00Z,0,0,900,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100') == \ + 'oldest_wait_seconds=900 limit=900' ]] for failed_pressure_row in \ '2026-08-01T00:00:00Z,0,0,901,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' \