From 132cda35d8b83d3be0ad516864042122822a3461 Mon Sep 17 00:00:00 2001 From: wangbo Date: Sat, 1 Aug 2026 01:51:44 +0800 Subject: [PATCH] =?UTF-8?q?fix(acceptance):=20=E9=9A=94=E7=A6=BB=E6=8E=A7?= =?UTF-8?q?=E5=88=B6=E9=9D=A2=E6=8A=96=E5=8A=A8=E4=B8=8E=E7=A7=9F=E7=BA=A6?= =?UTF-8?q?=E7=9E=AC=E6=80=81=E6=95=85=E9=9A=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 线上 P24 验收暴露出高频 kubectl exec 放大 K3s API 压力、门禁查询挤占关键连接池，以及 PostgreSQL 锁超时被误判为租约所有权丢失。 本次合并验收身份查询、在租约有效期内重试瞬态续期错误、修复人工审核残留 attempt，并增加滚动后 etcd 稳定窗口、节点直连指标和双站独立报告。 验证：Go 全量测试、go vet、聚焦 race、gofmt、迁移安全检查、bash -n、ShellCheck、manual release test。 --- .../internal/httpapi/acceptance_handlers.go | 107 ++++++++- .../httpapi/acceptance_handlers_test.go | 76 ++++++ apps/api/internal/httpapi/server.go | 3 + apps/api/internal/runner/execution_lease.go | 25 +- .../internal/runner/execution_lease_test.go | 33 +++ apps/api/internal/runner/service.go | 44 +++- .../store/billing_v2_integration_test.go | 11 + apps/api/internal/store/tasks_runtime.go | 13 + docs/operations/production-acceptance.md | 15 +- scripts/cluster/run-production-acceptance.sh | 224 +++++++++++++++--- 10 files changed, 495 insertions(+), 56 deletions(-) create mode 100644 apps/api/internal/runner/execution_lease_test.go diff --git a/apps/api/internal/httpapi/acceptance_handlers.go b/apps/api/internal/httpapi/acceptance_handlers.go index d368a3e..d1c842f 100644 --- a/apps/api/internal/httpapi/acceptance_handlers.go +++ b/apps/api/internal/httpapi/acceptance_handlers.go @@ -1,21 +1,43 @@ package httpapi import ( + "context" + "crypto/sha256" "encoding/json" "errors" "net/http" "strings" + "time" "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) const ( - acceptanceRunHeader = "X-EasyAI-Acceptance-Run" - acceptanceTokenHeader = "X-EasyAI-Acceptance-Token" - acceptanceUpstreamHeader = "X-EasyAI-Acceptance-Upstream" + acceptanceRunHeader = "X-EasyAI-Acceptance-Run" + acceptanceTokenHeader = "X-EasyAI-Acceptance-Token" + acceptanceUpstreamHeader = "X-EasyAI-Acceptance-Upstream" + taskTrafficAuthorizationCacheTTL = 5 * time.Second ) +type taskTrafficAuthorizationCacheKey struct { + runID string + apiKeyID string + userID string + tokenHash [sha256.Size]byte +} + +type taskTrafficAuthorizationCacheEntry struct { + runID string + expiresAt time.Time +} + +type taskTrafficAuthorizationCall struct { + done chan struct{} + runID string + err error +} + type taskTrafficAdmission struct { RunMode string AcceptanceRunID string @@ -44,12 +66,7 @@ func (e *taskTrafficError) ErrorCode() string { } func (s *Server) authorizeTaskTraffic(r *http.Request, user *auth.User) (taskTrafficAdmission, error) { - runID, err := s.acceptanceStore().AuthorizeAcceptanceTask( - r.Context(), - r.Header.Get(acceptanceRunHeader), - r.Header.Get(acceptanceTokenHeader), - user, - ) + runID, err := s.authorizeAcceptanceTask(r.Context(), r.Header.Get(acceptanceRunHeader), r.Header.Get(acceptanceTokenHeader), user) if err != nil { switch { case errors.Is(err, store.ErrProductionTrafficPaused): @@ -84,6 +101,78 @@ func (s *Server) authorizeTaskTraffic(r *http.Request, user *auth.User) (taskTra return taskTrafficAdmission{RunMode: "production"}, nil } +// authorizeAcceptanceTask collapses concurrent validation requests for the same +// participant into one critical-pool lookup and briefly caches only successful +// validation grants. Live requests and incomplete acceptance credentials always +// consult PostgreSQL so a validation transition remains fail-closed. +func (s *Server) authorizeAcceptanceTask(ctx context.Context, runID string, token string, user *auth.User) (string, error) { + runID = strings.TrimSpace(runID) + token = strings.TrimSpace(token) + if runID == "" || token == "" || user == nil || + strings.TrimSpace(user.APIKeyID) == "" || strings.TrimSpace(user.ID) == "" { + return s.acceptanceStore().AuthorizeAcceptanceTask(ctx, runID, token, user) + } + key := taskTrafficAuthorizationCacheKey{ + runID: runID, apiKeyID: strings.TrimSpace(user.APIKeyID), userID: strings.TrimSpace(user.ID), + tokenHash: sha256.Sum256([]byte(token)), + } + return s.cachedTaskTrafficAuthorization(ctx, key, taskTrafficAuthorizationCacheTTL, func() (string, error) { + return s.acceptanceStore().AuthorizeAcceptanceTask(ctx, runID, token, user) + }) +} + +func (s *Server) cachedTaskTrafficAuthorization( + ctx context.Context, + key taskTrafficAuthorizationCacheKey, + ttl time.Duration, + load func() (string, error), +) (string, error) { + now := time.Now() + s.taskTrafficCacheMu.Lock() + if entry, ok := s.taskTrafficCache[key]; ok { + if now.Before(entry.expiresAt) { + s.taskTrafficCacheMu.Unlock() + return entry.runID, nil + } + delete(s.taskTrafficCache, key) + } + if call, ok := s.taskTrafficInflight[key]; ok { + s.taskTrafficCacheMu.Unlock() + select { + case <-call.done: + return call.runID, call.err + case <-ctx.Done(): + return "", ctx.Err() + } + } + if s.taskTrafficInflight == nil { + s.taskTrafficInflight = make(map[taskTrafficAuthorizationCacheKey]*taskTrafficAuthorizationCall) + } + call := &taskTrafficAuthorizationCall{done: make(chan struct{})} + s.taskTrafficInflight[key] = call + s.taskTrafficCacheMu.Unlock() + + call.runID, call.err = load() + + s.taskTrafficCacheMu.Lock() + delete(s.taskTrafficInflight, key) + if call.err == nil && call.runID != "" { + if s.taskTrafficCache == nil { + s.taskTrafficCache = make(map[taskTrafficAuthorizationCacheKey]taskTrafficAuthorizationCacheEntry) + } + callTTL := ttl + if callTTL <= 0 { + callTTL = taskTrafficAuthorizationCacheTTL + } + s.taskTrafficCache[key] = taskTrafficAuthorizationCacheEntry{ + runID: call.runID, expiresAt: time.Now().Add(callTTL), + } + } + close(call.done) + s.taskTrafficCacheMu.Unlock() + return call.runID, call.err +} + func (s *Server) admittedTaskRunMode(admission taskTrafficAdmission, body map[string]any) (string, error) { if admission.RunMode == "acceptance" || admission.RunMode == "acceptance_canary" { return admission.RunMode, nil diff --git a/apps/api/internal/httpapi/acceptance_handlers_test.go b/apps/api/internal/httpapi/acceptance_handlers_test.go index 0d53d6a..740f94e 100644 --- a/apps/api/internal/httpapi/acceptance_handlers_test.go +++ b/apps/api/internal/httpapi/acceptance_handlers_test.go @@ -1,8 +1,12 @@ package httpapi import ( + "context" "errors" + "sync" + "sync/atomic" "testing" + "time" "github.com/easyai/easyai-ai-gateway/apps/api/internal/config" ) @@ -25,3 +29,75 @@ func TestAdmittedTaskRunModeReservesAcceptanceAndProductionSimulation(t *testing } } } + +func TestCachedTaskTrafficAuthorizationCollapsesConcurrentLoads(t *testing.T) { + server := &Server{} + key := taskTrafficAuthorizationCacheKey{runID: "run", apiKeyID: "key", userID: "user"} + var loads atomic.Int32 + start := make(chan struct{}) + const callers = 64 + var wg sync.WaitGroup + errCh := make(chan error, callers) + for range callers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + runID, err := server.cachedTaskTrafficAuthorization( + context.Background(), key, time.Minute, + func() (string, error) { + loads.Add(1) + time.Sleep(10 * time.Millisecond) + return "run", nil + }, + ) + if err != nil || runID != "run" { + errCh <- errors.New("unexpected cached authorization result") + } + }() + } + close(start) + wg.Wait() + close(errCh) + for err := range errCh { + t.Fatal(err) + } + if got := loads.Load(); got != 1 { + t.Fatalf("authorization loads=%d, want 1", got) + } + + if _, err := server.cachedTaskTrafficAuthorization( + context.Background(), key, time.Minute, + func() (string, error) { + loads.Add(1) + return "run", nil + }, + ); err != nil { + t.Fatal(err) + } + if got := loads.Load(); got != 1 { + t.Fatalf("cached authorization loads=%d, want 1", got) + } +} + +func TestCachedTaskTrafficAuthorizationDoesNotCacheFailures(t *testing.T) { + server := &Server{} + key := taskTrafficAuthorizationCacheKey{runID: "run", apiKeyID: "key", userID: "user"} + wantErr := errors.New("database unavailable") + loads := 0 + for range 2 { + _, err := server.cachedTaskTrafficAuthorization( + context.Background(), key, time.Minute, + func() (string, error) { + loads++ + return "", wantErr + }, + ) + if !errors.Is(err, wantErr) { + t.Fatalf("authorization error=%v, want %v", err, wantErr) + } + } + if loads != 2 { + t.Fatalf("failed authorization loads=%d, want 2", loads) + } +} diff --git a/apps/api/internal/httpapi/server.go b/apps/api/internal/httpapi/server.go index 9861bfc..a54991c 100644 --- a/apps/api/internal/httpapi/server.go +++ b/apps/api/internal/httpapi/server.go @@ -51,6 +51,9 @@ type Server struct { identityTestCookieSecure bool identityTestBrowserEnabled bool billingMetrics *ssfreceiver.Metrics + taskTrafficCacheMu sync.Mutex + taskTrafficCache map[taskTrafficAuthorizationCacheKey]taskTrafficAuthorizationCacheEntry + taskTrafficInflight map[taskTrafficAuthorizationCacheKey]*taskTrafficAuthorizationCall } type oidcPublicClient interface { diff --git a/apps/api/internal/runner/execution_lease.go b/apps/api/internal/runner/execution_lease.go index 70f08be..01af956 100644 --- a/apps/api/internal/runner/execution_lease.go +++ b/apps/api/internal/runner/execution_lease.go @@ -11,24 +11,37 @@ import ( const ( taskExecutionLeaseTTL = 5 * time.Minute taskExecutionRenewInterval = 30 * time.Second + leaseRenewalRetryInterval = 2 * time.Second ) func (s *Service) renewTaskExecutionLease(ctx context.Context, cancel context.CancelFunc, taskID string, executionToken string) { - ticker := time.NewTicker(taskExecutionRenewInterval) - defer ticker.Stop() + timer := time.NewTimer(taskExecutionRenewInterval) + defer timer.Stop() + leaseDeadline := time.Now().Add(taskExecutionLeaseTTL) for { select { case <-ctx.Done(): return - case <-ticker.C: + case <-timer.C: if err := s.coordinationStore.RenewTaskExecutionLease(ctx, taskID, executionToken, taskExecutionLeaseTTL); err != nil { if errors.Is(err, store.ErrTaskExecutionFinished) { return } - s.logger.Warn("task execution lease lost", "taskID", taskID, "error_category", "task_execution_lease_lost") - cancel() - return + if taskExecutionLeaseRenewalIsFatal(err, time.Now(), leaseDeadline) { + s.logger.Warn("task execution lease lost", "taskID", taskID, "error_category", "task_execution_lease_lost", "error", err) + cancel() + return + } + s.logger.Warn("task execution lease renewal failed; retrying before lease expiry", "taskID", taskID, "error_category", "task_execution_lease_renewal_transient", "error", err) + timer.Reset(leaseRenewalRetryInterval) + continue } + leaseDeadline = time.Now().Add(taskExecutionLeaseTTL) + timer.Reset(taskExecutionRenewInterval) } } } + +func taskExecutionLeaseRenewalIsFatal(err error, now time.Time, leaseDeadline time.Time) bool { + return errors.Is(err, store.ErrTaskExecutionLeaseLost) || !now.Before(leaseDeadline) +} diff --git a/apps/api/internal/runner/execution_lease_test.go b/apps/api/internal/runner/execution_lease_test.go new file mode 100644 index 0000000..6a304d9 --- /dev/null +++ b/apps/api/internal/runner/execution_lease_test.go @@ -0,0 +1,33 @@ +package runner + +import ( + "errors" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestLeaseRenewalFatalityDistinguishesTransientDatabaseErrors(t *testing.T) { + now := time.Now() + deadline := now.Add(time.Minute) + transient := errors.New("lock timeout") + if taskExecutionLeaseRenewalIsFatal(transient, now, deadline) { + t.Fatal("transient task lease renewal error must be retried before expiry") + } + if concurrencyLeaseRenewalIsFatal(transient, now, deadline) { + t.Fatal("transient concurrency lease renewal error must be retried before expiry") + } + if !taskExecutionLeaseRenewalIsFatal(store.ErrTaskExecutionLeaseLost, now, deadline) { + t.Fatal("confirmed task lease ownership loss must be fatal") + } + if !concurrencyLeaseRenewalIsFatal(store.ErrConcurrencyLeaseLost, now, deadline) { + t.Fatal("confirmed concurrency lease ownership loss must be fatal") + } + if !taskExecutionLeaseRenewalIsFatal(transient, deadline, deadline) { + t.Fatal("task lease renewal error at expiry must be fatal") + } + if !concurrencyLeaseRenewalIsFatal(transient, deadline, deadline) { + t.Fatal("concurrency lease renewal error at expiry must be fatal") + } +} diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index 1c59203..9c66131 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -2107,32 +2107,48 @@ func (s *Service) startConcurrencyLeaseRenewal(ctx context.Context, taskID strin runCtx, cancelRun := context.WithCancel(ctx) renewCtx, cancelRenew := context.WithCancel(ctx) done := make(chan error, 1) + minimumTTL := 120 * time.Second + for _, lease := range leases { + if lease.TTL > 0 && lease.TTL < minimumTTL { + minimumTTL = lease.TTL + } + } go func() { - ticker := time.NewTicker(interval) - defer ticker.Stop() + timer := time.NewTimer(interval) + defer timer.Stop() + leaseDeadline := time.Now().Add(minimumTTL) for { select { case <-renewCtx.Done(): done <- nil return - case <-ticker.C: + case <-timer.C: if err := s.coordinationStore.RenewConcurrencyLeases(renewCtx, leases); err != nil { if renewCtx.Err() != nil { done <- nil return } - outcome := "failure" - if errors.Is(err, store.ErrConcurrencyLeaseLost) { - outcome = "lost" + if concurrencyLeaseRenewalIsFatal(err, time.Now(), leaseDeadline) { + outcome := "failure" + if errors.Is(err, store.ErrConcurrencyLeaseLost) { + outcome = "lost" + } + s.observeConcurrencyLeaseRenewal(outcome) + s.logger.Error("concurrency lease renewal failed; cancelling upstream execution", + "taskID", taskID, "leaseCount", len(leases), "outcome", outcome, "error", err) + done <- err + cancelRun() + return } - s.observeConcurrencyLeaseRenewal(outcome) - s.logger.Error("concurrency lease renewal failed; cancelling upstream execution", - "taskID", taskID, "leaseCount", len(leases), "outcome", outcome, "error", err) - done <- err - cancelRun() - return + s.observeConcurrencyLeaseRenewal("failure") + s.logger.Warn("concurrency lease renewal failed; retrying before lease expiry", + "taskID", taskID, "leaseCount", len(leases), "outcome", "transient", "error", err) + timer.Reset(leaseRenewalRetryInterval) + continue } s.observeConcurrencyLeaseRenewal("success") + leaseDeadline = time.Now().Add(minimumTTL) + timer.Reset(interval) } } }() @@ -2144,6 +2160,10 @@ func (s *Service) startConcurrencyLeaseRenewal(ctx context.Context, taskID strin } } +func concurrencyLeaseRenewalIsFatal(err error, now time.Time, leaseDeadline time.Time) bool { + return errors.Is(err, store.ErrConcurrencyLeaseLost) || !now.Before(leaseDeadline) +} + func (s *Service) observeConcurrencyLeaseRenewal(outcome string) { observer, ok := s.billingMetrics.(interface { ObserveConcurrencyLeaseRenewal(string) diff --git a/apps/api/internal/store/billing_v2_integration_test.go b/apps/api/internal/store/billing_v2_integration_test.go index 2651118..3651dfb 100644 --- a/apps/api/internal/store/billing_v2_integration_test.go +++ b/apps/api/internal/store/billing_v2_integration_test.go @@ -154,6 +154,17 @@ func TestExpiredExecutionLeaseDoesNotReplayAmbiguousUpstreamSubmission(t *testin if review.Status != "failed" || review.BillingStatus != "manual_review" || review.ErrorCode != "upstream_submission_unknown" { t.Fatalf("manual review task=%+v", review) } + var attemptStatus string + var attemptErrorCode string + if err := db.pool.QueryRow(ctx, ` +SELECT status, COALESCE(error_code, '') +FROM gateway_task_attempts +WHERE id=$1::uuid`, attemptID).Scan(&attemptStatus, &attemptErrorCode); err != nil { + t.Fatal(err) + } + if attemptStatus != "failed" || attemptErrorCode != "upstream_submission_unknown" { + t.Fatalf("manual review attempt status=%s error=%s", attemptStatus, attemptErrorCode) + } var outboxStatus string var outboxAction string var reviewReason string diff --git a/apps/api/internal/store/tasks_runtime.go b/apps/api/internal/store/tasks_runtime.go index b23748a..d3601c0 100644 --- a/apps/api/internal/store/tasks_runtime.go +++ b/apps/api/internal/store/tasks_runtime.go @@ -388,6 +388,19 @@ func markTaskExecutionManualReviewTx( hasGatewayUser bool, ) error { if _, err := tx.Exec(ctx, ` +UPDATE gateway_task_attempts +SET status = 'failed', + retryable = false, + error_code = 'upstream_submission_unknown', + error_message = 'upstream submission result is unknown', + finished_at = COALESCE(finished_at, now()), + upstream_submission_updated_at = now() +WHERE task_id = $1::uuid + AND status = 'running' + AND upstream_submission_status IN ('submitting', 'response_received')`, taskID); err != nil { + return err + } + if _, err := tx.Exec(ctx, ` UPDATE gateway_tasks SET status = 'failed', billing_status = CASE WHEN $2 THEN 'manual_review' ELSE 'not_required' END, diff --git a/docs/operations/production-acceptance.md b/docs/operations/production-acceptance.md index 5c029cb..52f63aa 100644 --- a/docs/operations/production-acceptance.md +++ b/docs/operations/production-acceptance.md @@ -110,6 +110,15 @@ API Pod 使用独立发布配置 `AI_GATEWAY_API_DATABASE_MAX_CONNS`，生产同 每档依次运行全部五个模拟 Profile 三次。失败时恢复上一个已完整通过的档位，流量保持 `validation`，不会自动放开正式请求。 +每次容量配置滚动后不会立即造压测流量。脚本先等待默认 60 秒的控制面稳定窗口，要求三地 +`readyz/etcd`、CNPG 同步副本和节点内存正常，窗口内没有 API Server handler timeout、 +etcd peer I/O timeout、WAL receiver timeout 或 CNPG 控制面连接异常；同时按 etcd counter +增量计算各节点 backend commit 和 WAL fsync 平均延迟。默认门槛分别为 50 ms 和 15 ms， +可通过 `AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS`、 +`AI_GATEWAY_ACCEPTANCE_ETCD_BACKEND_COMMIT_MAX_MS`、 +`AI_GATEWAY_ACCEPTANCE_ETCD_WAL_FSYNC_MAX_MS` 调整。最长等待 240 秒仍不稳定时直接阻断， +不会把发布滚动造成的控制面抖动与业务容量混在同一次负载里。 + 常规容量调整只需修改服务器上的私有发布配置并执行获授权的滚动命令，不要求源码改动或 新镜像： @@ -142,7 +151,9 @@ PNG 并通过验收 Key 上传到 Gateway。生产快照默认在已部署的 AP 压测器，校验 SHA-256 后以 `0600` 临时环境分别运行在宁波、香港宿主机的 K3s 之外；两个 进程同时只访问本站 TLS 入口，并按奇偶全局序号分担请求。这样避免本机上行带宽成为容量 瓶颈，同时仍覆盖正式 TLS/Nginx 入口。站点报告按请求数相加、耗时取较慢站点、p50/p95/p99 -取两站较差值进行保守合并。进程、私有环境和远端部分报告在 Run 退出时按精确路径清理。 +取两站较差值进行保守合并。宁波、香港的脱敏部分报告会分别保存在 Run 目录，远端进程和 +私有环境在 Run 退出时按精确路径清理。压力采样直接从节点访问 Pod metrics，不再反复创建 +`kubectl exec` 流，避免验收监控本身放大 K3s API Server 和 etcd 压力。 ```bash scripts/cluster/run-production-acceptance.sh \ @@ -166,7 +177,7 @@ scripts/cluster/run-production-acceptance.sh \ 2. 准备专属验收用户组、身份分片、独立钱包与候选访问规则。 3. 部署 digest 固定且仅集群内可访问的协议模拟器。 4. 创建 Run、切换 `validation`、等待已有正式任务排空。 -5. 按 P24、P28、P32 执行三轮负载和 Worker 强杀。 +5. 每档滚动后先通过控制面稳定窗口，再按 P24、P28、P32 执行三轮负载和 Worker 强杀。 6. 执行两条真实小流量请求。 7. 校验任务、账务、回调、队列、连接池、RSS、节点内存、数据库同步、六向 WireGuard 和 公网健康。 diff --git a/scripts/cluster/run-production-acceptance.sh b/scripts/cluster/run-production-acceptance.sh index 80b5f5e..118dc84 100755 --- a/scripts/cluster/run-production-acceptance.sh +++ b/scripts/cluster/run-production-acceptance.sh @@ -43,6 +43,9 @@ Optional model overrides (otherwise selected from current production candidates) AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS AI_GATEWAY_ACCEPTANCE_API_MEDIA_REQUEST_CONCURRENCY AI_GATEWAY_ACCEPTANCE_NODE_MEMORY_PRECONDITION_PERCENT + AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS + AI_GATEWAY_ACCEPTANCE_ETCD_BACKEND_COMMIT_MAX_MS + AI_GATEWAY_ACCEPTANCE_ETCD_WAL_FSYNC_MAX_MS EOF } @@ -90,6 +93,10 @@ require_commands curl git go jq node openssl sed shasum : "${AI_GATEWAY_ACCEPTANCE_WORKER_MEMORY_REQUEST_MIB:=1536}" : "${AI_GATEWAY_ACCEPTANCE_WORKER_CPU_REQUEST_MILLICORES:=500}" : "${AI_GATEWAY_ACCEPTANCE_NODE_MEMORY_PRECONDITION_PERCENT:=65}" +: "${AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS:=60}" +: "${AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_TIMEOUT_SECONDS:=240}" +: "${AI_GATEWAY_ACCEPTANCE_ETCD_BACKEND_COMMIT_MAX_MS:=50}" +: "${AI_GATEWAY_ACCEPTANCE_ETCD_WAL_FSYNC_MAX_MS:=15}" : "${AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO:=2}" : "${AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG:=2}" : "${AI_GATEWAY_ACCEPTANCE_SOAK_DURATION:=2h}" @@ -146,6 +153,20 @@ fi echo 'AI_GATEWAY_ACCEPTANCE_NODE_MEMORY_PRECONDITION_PERCENT must be between 50 and 79' >&2 exit 1 } +[[ $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS =~ ^[1-9][0-9]*$ && + $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS -ge 30 && + $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS -le 300 && + $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_TIMEOUT_SECONDS =~ ^[1-9][0-9]*$ && + $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_TIMEOUT_SECONDS -ge $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS && + $AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_TIMEOUT_SECONDS -le 900 ]] || { + echo 'post-rollout stability window must be 30..300 seconds and timeout must be window..900 seconds' >&2 + exit 1 +} +[[ $AI_GATEWAY_ACCEPTANCE_ETCD_BACKEND_COMMIT_MAX_MS =~ ^[1-9][0-9]*$ && + $AI_GATEWAY_ACCEPTANCE_ETCD_WAL_FSYNC_MAX_MS =~ ^[1-9][0-9]*$ ]] || { + echo 'etcd latency gates must be positive integer milliseconds' >&2 + exit 1 +} [[ $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO =~ ^[1-9][0-9]*$ && $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_HONGKONG =~ ^[1-9][0-9]*$ && $AI_GATEWAY_ACCEPTANCE_AUTOSCALING_MAX_REPLICAS_NINGBO -le 16 && @@ -244,6 +265,7 @@ video_stable_throughput= video_admitted_throughput= certified_max_replicas_ningbo=1 certified_max_replicas_hongkong=1 +runtime_observation_started_at= cleanup() { local status=$? @@ -1443,6 +1465,125 @@ apply_capacity_profile() { active_profile=$profile } +capture_etcd_latency_counters() { + local output=$1 + local site host metrics backend_sum backend_count wal_sum wal_count + echo 'site,backend_sum,backend_count,wal_sum,wal_count' >"$output" + for site in ningbo hongkong losangeles; do + case $site in + ningbo) host=$CLUSTER_NINGBO_HOST ;; + hongkong) host=$CLUSTER_HONGKONG_HOST ;; + losangeles) host=$CLUSTER_LOS_ANGELES_HOST ;; + esac + metrics=$(cluster_ssh "$host" 'curl -fsS --max-time 5 http://127.0.0.1:2381/metrics') || return 1 + backend_sum=$(awk '$1=="etcd_disk_backend_commit_duration_seconds_sum" {print $2}' <<<"$metrics") + backend_count=$(awk '$1=="etcd_disk_backend_commit_duration_seconds_count" {print $2}' <<<"$metrics") + wal_sum=$(awk '$1=="etcd_disk_wal_fsync_duration_seconds_sum" {print $2}' <<<"$metrics") + wal_count=$(awk '$1=="etcd_disk_wal_fsync_duration_seconds_count" {print $2}' <<<"$metrics") + [[ $backend_sum =~ ^[0-9.]+$ && $backend_count =~ ^[0-9]+$ && + $wal_sum =~ ^[0-9.]+$ && $wal_count =~ ^[0-9]+$ ]] || return 1 + echo "$site,$backend_sum,$backend_count,$wal_sum,$wal_count" >>"$output" + done +} + +control_plane_hard_errors_since() { + local since=$1 + local host total=0 count pod + [[ $since =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$ ]] + for host in "$CLUSTER_NINGBO_HOST" "$CLUSTER_HONGKONG_HOST" "$CLUSTER_LOS_ANGELES_HOST"; do + count=$(cluster_ssh "$host" \ + "journalctl -u k3s --since '$since' --no-pager 2>/dev/null | awk 'BEGIN {IGNORECASE=1} /http: Handler timeout/ || /rejected connection on peer endpoint.*(timeout|reset)/ || /etcdserver: request timed out/ || /leader changed/ {count++} END {print count+0}'") || return 1 + [[ $count =~ ^[0-9]+$ ]] || return 1 + total=$((total + count)) + done + while read -r pod; do + [[ -n $pod ]] || continue + count=$(remote_kubectl logs "$pod" -n "$namespace" -c postgres --since-time="$since" 2>/dev/null | + awk 'BEGIN {IGNORECASE=1} + /API server connectivity issue/ || + /Instance connectivity error/ || + /terminating walreceiver due to timeout/ || + /synchronous commit.*cancel/ {count++} + END {print count+0}') || return 1 + [[ $count =~ ^[0-9]+$ ]] || return 1 + total=$((total + count)) + done < <(remote_kubectl get pods -n "$namespace" \ + -l 'cnpg.io/cluster=easyai-postgres' \ + -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}') + echo "$total" +} + +evaluate_etcd_latency_window() { + local before=$1 + local after=$2 + local output=$3 + echo 'site,backend_commit_average_ms,wal_fsync_average_ms' >"$output" + awk -F',' \ + -v backend_limit="$AI_GATEWAY_ACCEPTANCE_ETCD_BACKEND_COMMIT_MAX_MS" \ + -v wal_limit="$AI_GATEWAY_ACCEPTANCE_ETCD_WAL_FSYNC_MAX_MS" ' + NR==FNR { + if (FNR>1) { + backend_sum[$1]=$2; backend_count[$1]=$3 + wal_sum[$1]=$4; wal_count[$1]=$5 + } + next + } + FNR==1 {next} + { + backend_delta_count=$3-backend_count[$1] + wal_delta_count=$5-wal_count[$1] + if (backend_delta_count<=0 || wal_delta_count<=0) {failed=1; next} + backend_ms=($2-backend_sum[$1])*1000/backend_delta_count + wal_ms=($4-wal_sum[$1])*1000/wal_delta_count + printf "%s,%.3f,%.3f\n", $1, backend_ms, wal_ms + if (backend_ms>=backend_limit || wal_ms>=wal_limit) failed=1 + } + END {exit failed} + ' "$before" "$after" >>"$output" +} + +wait_for_post_rollout_stability() { + local profile=$1 + local profile_slug deadline attempt=0 since before after latency_report errors + profile_slug=$(printf '%s' "$profile" | tr '[:upper:]' '[:lower:]') + latency_report=$report_root/$profile_slug-control-plane-stability.csv + deadline=$((SECONDS + AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_TIMEOUT_SECONDS)) + while (( SECONDS + AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS <= deadline )); do + attempt=$((attempt + 1)) + since=$(date -u '+%Y-%m-%dT%H:%M:%SZ') + before=$temporary_root/etcd-before-$profile_slug-$attempt.csv + after=$temporary_root/etcd-after-$profile_slug-$attempt.csv + capture_etcd_latency_counters "$before" || continue + sleep "$AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS" + capture_etcd_latency_counters "$after" || continue + errors=$(control_plane_hard_errors_since "$since" || echo 1) + if [[ $errors == 0 ]] && evaluate_etcd_latency_window "$before" "$after" "$latency_report" && + verify_control_plane_ready_now; then + runtime_observation_started_at=$(date -u '+%Y-%m-%dT%H:%M:%SZ') + echo "post_rollout_stability=PASS profile=$profile window_seconds=$AI_GATEWAY_ACCEPTANCE_POST_ROLLOUT_STABILITY_SECONDS hard_errors=0" + return 0 + fi + echo "post-rollout control plane not stable: profile=$profile attempt=$attempt hard_errors=${errors:-unknown}" >&2 + done + return 1 +} + +verify_control_plane_ready_now() { + local host ready_output sync_state memory_percent + for host in "$CLUSTER_NINGBO_HOST" "$CLUSTER_HONGKONG_HOST" "$CLUSTER_LOS_ANGELES_HOST"; do + ready_output=$(cluster_ssh "$host" "k3s kubectl get --raw='/readyz?verbose'") + grep -Fq '[+]etcd ok' <<<"$ready_output" || return 1 + done + [[ $(remote_kubectl get cluster easyai-postgres -n "$namespace" -o 'jsonpath={.status.readyInstances}') == 2 ]] || return 1 + sync_state=$(database_query "SELECT COALESCE(string_agg(sync_state,','),'') FROM pg_stat_replication;") + [[ ",$sync_state," == *,sync,* || ",$sync_state," == *,quorum,* ]] || return 1 + while read -r _node _cpu _cpu_percent _memory memory_percent; do + memory_percent=${memory_percent%\%} + [[ $memory_percent =~ ^[0-9]+$ ]] || return 1 + (( memory_percent < 80 )) || return 1 + done < <(remote_kubectl top nodes --no-headers) +} + capacity_controller_status() { local pod status while read -r pod; do @@ -2078,7 +2219,10 @@ run_distributed_load_profile() { 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" + install -m 0600 "$ningbo_report" "${report_path%.json}-ningbo.json" + install -m 0600 "$hongkong_report" "${report_path%.json}-hongkong.json" + merge_remote_load_reports "$profile" \ + "${report_path%.json}-ningbo.json" "${report_path%.json}-hongkong.json" "$report_path" (( ningbo_status == 0 && hongkong_status == 0 )) || return 1 jq -e '.secretSafe == true and .passed == true' "$report_path" >/dev/null } @@ -2191,40 +2335,76 @@ verify_cluster_links() { } verify_control_plane_and_recent_logs() { - local host ready_output workload pod error_count + local host ready_output workload pod error_count log_window + log_window=--since=10m + if [[ -n $runtime_observation_started_at ]]; then + log_window=--since-time=$runtime_observation_started_at + fi for host in "$CLUSTER_NINGBO_HOST" "$CLUSTER_HONGKONG_HOST" "$CLUSTER_LOS_ANGELES_HOST"; do ready_output=$(cluster_ssh "$host" "k3s kubectl get --raw='/readyz?verbose'") - grep -Fq '[+]etcd ok' <<<"$ready_output" + grep -Fq '[+]etcd ok' <<<"$ready_output" || return 1 + if [[ -n $runtime_observation_started_at ]]; then + error_count=$(cluster_ssh "$host" \ + "journalctl -u k3s --since '$runtime_observation_started_at' --no-pager 2>/dev/null | awk 'BEGIN {IGNORECASE=1} /http: Handler timeout/ || /rejected connection on peer endpoint.*(timeout|reset)/ || /etcdserver: request timed out/ || /leader changed/ {count++} END {print count+0}'") + [[ $error_count == 0 ]] || return 1 + fi done for workload in api worker; do while read -r pod; do - error_count=$(remote_kubectl logs "$pod" -n "$namespace" --since=10m | + error_count=$(remote_kubectl logs "$pod" -n "$namespace" "$log_window" | awk 'BEGIN {IGNORECASE=1} /postgres_unavailable/ || /postgres readiness.*(fail|unavailable|error)/ || /leadership elector.*(error|fail)/ || - /concurrency lease.*(renew.*(error|fail)|lost)/ {count++} + /concurrency lease.*(renew.*(error|fail)|lost)/ || + /task execution lease (lost|renewal failed)/ || + /upstream submission requires manual review/ {count++} END {print count+0}') - [[ $error_count == 0 ]] + [[ $error_count == 0 ]] || return 1 done < <(remote_kubectl get pods -n "$namespace" \ -l "app.kubernetes.io/name=easyai-$workload" \ -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}') done while read -r pod; do - error_count=$(remote_kubectl logs "$pod" -n "$namespace" -c postgres --since=10m | + error_count=$(remote_kubectl logs "$pod" -n "$namespace" -c postgres "$log_window" | awk 'BEGIN {IGNORECASE=1} /apply request took too long/ || /dial tcp.*6443/ || + /API server connectivity issue/ || + /Instance connectivity error/ || + /terminating walreceiver due to timeout/ || /synchronous commit.*cancel/ || /canceling statement due to user request/ || + /canceling statement due to lock timeout/ || /(liveness|readiness).*fail/ {count++} END {print count+0}') - [[ $error_count == 0 ]] + [[ $error_count == 0 ]] || return 1 done < <(remote_kubectl get pods -n "$namespace" \ -l 'cnpg.io/cluster=easyai-postgres' \ -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}') } +workload_metrics_for_site() { + local workload=$1 + local site=$2 + local host pod_ip + [[ $workload == api || $workload == worker ]] + [[ $site == ningbo || $site == hongkong ]] + if [[ $site == ningbo ]]; then + host=$CLUSTER_NINGBO_HOST + else + host=$CLUSTER_HONGKONG_HOST + fi + pod_ip=$(remote_kubectl get pods -n "$namespace" \ + -l "app.kubernetes.io/name=easyai-$workload,easyai.io/site=$site" -o json | + jq -r '[.items[] + | select(.metadata.deletionTimestamp == null) + | select(any(.status.conditions[]?; .type=="Ready" and .status=="True")) + | .status.podIP][0] // empty') + [[ $pod_ip =~ ^[0-9a-fA-F:.]+$ ]] || return 1 + cluster_ssh "$host" "curl -fsS --max-time 5 http://$pod_ip:8088/metrics" +} + current_max_pod_memory_mib() { remote_kubectl top pods -n "$namespace" \ -l 'app.kubernetes.io/part-of=easyai-ai-gateway' --no-headers | @@ -2432,11 +2612,7 @@ WHERE status='active' local river_pool_max river_acquired river_idle local lease_failures lease_lost for site in ningbo hongkong; do - pod=$(remote_kubectl get pods -n "$namespace" \ - -l "app.kubernetes.io/name=easyai-api,easyai.io/site=$site" \ - -o 'jsonpath={.items[0].metadata.name}') - metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- \ - wget -qO- http://127.0.0.1:8088/metrics) + metrics=$(workload_metrics_for_site api "$site") pool_max=$(awk '$1=="easyai_gateway_postgres_pool_max_connections" {print $2}' <<<"$metrics") acquired=$(awk '$1=="easyai_gateway_postgres_pool_acquired_connections" {print $2}' <<<"$metrics") idle=$(awk '$1=="easyai_gateway_postgres_pool_idle_connections" {print $2}' <<<"$metrics") @@ -2458,10 +2634,7 @@ WHERE status='active' done for _ in 1 2 3 4 5 6; do for site in ningbo hongkong; do - pod=$(remote_kubectl get pods -n "$namespace" \ - -l "app.kubernetes.io/name=easyai-worker,easyai.io/site=$site" \ - -o 'jsonpath={.items[0].metadata.name}') - metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- wget -qO- http://127.0.0.1:8088/metrics) + metrics=$(workload_metrics_for_site worker "$site") desired=$(awk '$1=="easyai_gateway_async_worker_desired_capacity" {print $2}' <<<"$metrics") current=$(awk '$1=="easyai_gateway_async_worker_capacity" {print $2}' <<<"$metrics") global=$(awk '$1=="easyai_gateway_worker_global_capacity" {print $2}' <<<"$metrics") @@ -2508,7 +2681,7 @@ WHERE status='active' sample_pressure() { local output=$1 local stop_file=$2 - local database_state node_memory_percent pod_memory_mib worker_resources pool_state pod metrics + local database_state node_memory_percent pod_memory_mib worker_resources pool_state metrics local consecutive_failures=0 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" while [[ ! -f $stop_file ]]; do @@ -2536,15 +2709,7 @@ WHERE acceptance_run_id='$run_id'::uuid;"); then if ! pool_state=$( for workload in api worker; do for site in ningbo hongkong; do - pod=$(remote_kubectl get pods -n "$namespace" \ - -l "app.kubernetes.io/name=easyai-$workload,easyai.io/site=$site" -o json | - jq -r '[.items[] - | select(.metadata.deletionTimestamp == null) - | select(any(.status.conditions[]?; .type=="Ready" and .status=="True")) - | .metadata.name][0] // empty') || exit - [[ -n $pod ]] || continue - metrics=$(remote_kubectl exec -n "$namespace" "$pod" -- \ - wget -qO- http://127.0.0.1:8088/metrics) || exit + metrics=$(workload_metrics_for_site "$workload" "$site") || exit awk -v workload="$workload" ' $1=="easyai_gateway_postgres_pool_acquired_connections" {acquired=$2} $1=="easyai_gateway_postgres_pool_max_connections" {pool_max=$2} @@ -3132,6 +3297,11 @@ for profile in P24 P28 P32; do failure_reason="failed to apply $profile" break fi + if ! wait_for_post_rollout_stability "$profile"; then + failure_gate_id=post_rollout_control_plane_stability + failure_reason="$profile control plane did not stabilize after rollout" + break + fi profile_passed=true for repetition in 1 2 3; do if ! run_capacity_round "$profile" "$repetition"; then