fix(acceptance): 按隔离额度执行平台降级路由

让 acceptance 候选负载读取同一 Run ID 下的并发、RPM 和 TPM 计数器,并在单个平台模型额度不足时继续尝试下一候选,生产与 canary 额度作用域保持不变。\n\n补齐本地容量控制器对宁波和香港测试 Worker 的最小 RBAC,避免初次应用生产反亲和配置时发生滚动死锁,并将突发负载对齐到完整限流窗口。\n\n验证:\n- go test ./internal/runner ./internal/store -count=1\n- go vet ./internal/runner ./internal/store\n- bash -n scripts/acceptance/local-cluster.sh scripts/acceptance/provider-burst.sh scripts/acceptance/run-local-acceptance.sh\n- shellcheck -x scripts/acceptance/local-cluster.sh scripts/acceptance/provider-burst.sh scripts/acceptance/run-local-acceptance.sh\n- kubectl apply --dry-run=client -f deploy/kubernetes/local-acceptance/capacity-controller-rbac.yaml
This commit is contained in:
2026-08-03 12:06:09 +08:00
parent 8485f14bc4
commit d740e9f676
7 changed files with 67 additions and 4 deletions
+19
View File
@@ -103,6 +103,7 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding(
CacheAffinityKey: cacheAffinityKeys.Primary, CacheAffinityKey: cacheAffinityKeys.Primary,
CacheAffinityKeys: cacheAffinityKeys.Lookup, CacheAffinityKeys: cacheAffinityKeys.Lookup,
CacheAffinityPolicy: runnerPolicy.CacheAffinityPolicy, CacheAffinityPolicy: runnerPolicy.CacheAffinityPolicy,
QuotaScopePrefix: acceptanceQuotaScopePrefix(task),
}) })
if err == nil { if err == nil {
candidates, err = filterCandidatesByRequestedPlatform(candidates, body) candidates, err = filterCandidatesByRequestedPlatform(candidates, body)
@@ -120,6 +121,7 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding(
return taskAdmissionPlan{}, err return taskAdmissionPlan{}, err
} }
candidates, _ = pinCandidatesToTaskAdmission(candidates, admission) candidates, _ = pinCandidatesToTaskAdmission(candidates, admission)
var candidateRateLimitErr error
for _, candidate := range candidates { for _, candidate := range candidates {
available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID) available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID)
if availabilityErr != nil { if availabilityErr != nil {
@@ -151,6 +153,13 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding(
s.rateLimitReservations(ctx, user, candidate, body), s.rateLimitReservations(ctx, user, candidate, body),
) )
if err := s.store.CheckRateLimits(ctx, reservations); err != nil { if err := s.store.CheckRateLimits(ctx, reservations); err != nil {
var limitErr *store.RateLimitExceededError
if errors.As(err, &limitErr) && limitErr.ScopeType == "platform_model" {
if candidateRateLimitErr == nil {
candidateRateLimitErr = err
}
continue
}
return taskAdmissionPlan{}, err return taskAdmissionPlan{}, err
} }
return taskAdmissionPlan{ return taskAdmissionPlan{
@@ -162,6 +171,9 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding(
Eligible: true, Eligible: true,
}, nil }, nil
} }
if candidateRateLimitErr != nil {
return taskAdmissionPlan{}, candidateRateLimitErr
}
return taskAdmissionPlan{}, store.ErrNoModelCandidate return taskAdmissionPlan{}, store.ErrNoModelCandidate
} }
@@ -239,6 +251,13 @@ func acceptanceScopeKey(task store.GatewayTask, scopeKey string) string {
return "acceptance:" + runID + ":" + scopeKey return "acceptance:" + runID + ":" + scopeKey
} }
func acceptanceQuotaScopePrefix(task store.GatewayTask) string {
if task.RunMode != "acceptance" {
return ""
}
return acceptanceScopeKey(task, "")
}
func (s *Service) loadAsyncTaskAdmission(ctx context.Context, task store.GatewayTask) (*store.TaskAdmission, error) { func (s *Service) loadAsyncTaskAdmission(ctx context.Context, task store.GatewayTask) (*store.TaskAdmission, error) {
if !task.AsyncMode { if !task.AsyncMode {
return nil, nil return nil, nil
@@ -117,6 +117,19 @@ func TestAcceptanceInfrastructureReservationsIsolateEveryMetric(t *testing.T) {
} }
} }
func TestAcceptanceQuotaScopePrefixMatchesIsolatedCounters(t *testing.T) {
acceptanceTask := store.GatewayTask{RunMode: "acceptance", AcceptanceRunID: "run-1"}
if got := acceptanceQuotaScopePrefix(acceptanceTask); got != "acceptance:run-1:" {
t.Fatalf("acceptance quota scope prefix=%q", got)
}
if got := acceptanceQuotaScopePrefix(store.GatewayTask{RunMode: "acceptance_canary", AcceptanceRunID: "run-1"}); got != "" {
t.Fatalf("canary must use production quota scope, got %q", got)
}
if got := acceptanceQuotaScopePrefix(store.GatewayTask{RunMode: "production"}); got != "" {
t.Fatalf("production quota scope changed, got %q", got)
}
}
func TestAcceptanceAdmissionScopesLeaveProductionPolicyUnchanged(t *testing.T) { func TestAcceptanceAdmissionScopesLeaveProductionPolicyUnchanged(t *testing.T) {
input := []store.AdmissionScope{{ input := []store.AdmissionScope{{
ScopeType: "platform_model", ScopeType: "platform_model",
+1
View File
@@ -372,6 +372,7 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask,
CacheAffinityKey: cacheAffinityKeys.Primary, CacheAffinityKey: cacheAffinityKeys.Primary,
CacheAffinityKeys: cacheAffinityKeys.Lookup, CacheAffinityKeys: cacheAffinityKeys.Lookup,
CacheAffinityPolicy: runnerPolicy.CacheAffinityPolicy, CacheAffinityPolicy: runnerPolicy.CacheAffinityPolicy,
QuotaScopePrefix: acceptanceQuotaScopePrefix(task),
}) })
if err == nil { if err == nil {
candidates, err = filterCandidatesByRequestedPlatform(candidates, body) candidates, err = filterCandidatesByRequestedPlatform(candidates, body)
+6 -4
View File
@@ -17,6 +17,7 @@ type ListModelCandidatesOptions struct {
CacheAffinityKey string CacheAffinityKey string
CacheAffinityKeys []string CacheAffinityKeys []string
CacheAffinityPolicy map[string]any CacheAffinityPolicy map[string]any
QuotaScopePrefix string
} }
func (s *Store) ListModelCandidates(ctx context.Context, model string, modelType string, user *auth.User, options ...ListModelCandidatesOptions) ([]RuntimeModelCandidate, error) { func (s *Store) ListModelCandidates(ctx context.Context, model string, modelType string, user *auth.User, options ...ListModelCandidatesOptions) ([]RuntimeModelCandidate, error) {
@@ -102,7 +103,7 @@ LEFT JOIN (
AND released_at IS NULL AND released_at IS NULL
AND expires_at > statement_timestamp() AND expires_at > statement_timestamp()
GROUP BY scope_key GROUP BY scope_key
) con ON con.scope_key = m.id::text ) con ON con.scope_key = $5::text || m.id::text
LEFT JOIN ( LEFT JOIN (
SELECT queued_sources.platform_model_id, COUNT(DISTINCT queued_sources.task_id) AS waiting SELECT queued_sources.platform_model_id, COUNT(DISTINCT queued_sources.task_id) AS waiting
FROM ( FROM (
@@ -137,7 +138,7 @@ LEFT JOIN (
AND metric = 'rpm' AND metric = 'rpm'
AND reset_at > now() AND reset_at > now()
ORDER BY scope_key, window_start DESC ORDER BY scope_key, window_start DESC
) rpm ON rpm.scope_key = m.id::text ) rpm ON rpm.scope_key = $5::text || m.id::text
LEFT JOIN ( LEFT JOIN (
SELECT scope_key, SUM(used_value) AS used_value, SUM(reserved_value) AS reserved_value SELECT scope_key, SUM(used_value) AS used_value, SUM(reserved_value) AS reserved_value
FROM gateway_rate_limit_counters FROM gateway_rate_limit_counters
@@ -145,7 +146,7 @@ LEFT JOIN (
AND metric LIKE 'tpm%' AND metric LIKE 'tpm%'
AND reset_at > now() AND reset_at > now()
GROUP BY scope_key GROUP BY scope_key
) tpm ON tpm.scope_key = m.id::text ) tpm ON tpm.scope_key = $5::text || m.id::text
WHERE p.status = 'enabled' WHERE p.status = 'enabled'
AND p.deleted_at IS NULL AND p.deleted_at IS NULL
AND m.enabled = true AND m.enabled = true
@@ -170,7 +171,7 @@ WHERE p.status = 'enabled'
COALESCE(s.running_count, 0) ASC, COALESCE(s.running_count, 0) ASC,
COALESCE(s.waiting_count, 0) ASC, COALESCE(s.waiting_count, 0) ASC,
COALESCE(s.last_assigned_at, to_timestamp(0)) ASC, COALESCE(s.last_assigned_at, to_timestamp(0)) ASC,
m.created_at ASC`, exactModel, modelType, listOptions.CacheAffinityKeys, cacheAffinityStaleAfterSeconds(listOptions.CacheAffinityPolicy)) m.created_at ASC`, exactModel, modelType, listOptions.CacheAffinityKeys, cacheAffinityStaleAfterSeconds(listOptions.CacheAffinityPolicy), listOptions.QuotaScopePrefix)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -472,6 +473,7 @@ func normalizeListModelCandidatesOptions(modelType string, options ...ListModelC
} }
out := options[0] out := options[0]
out.CacheAffinityKey = strings.TrimSpace(out.CacheAffinityKey) out.CacheAffinityKey = strings.TrimSpace(out.CacheAffinityKey)
out.QuotaScopePrefix = strings.TrimSpace(out.QuotaScopePrefix)
out.CacheAffinityKeys = normalizedCacheAffinityKeys(out.CacheAffinityKey, out.CacheAffinityKeys) out.CacheAffinityKeys = normalizedCacheAffinityKeys(out.CacheAffinityKey, out.CacheAffinityKeys)
if len(out.CacheAffinityKeys) > 0 { if len(out.CacheAffinityKeys) > 0 {
out.CacheAffinityKey = out.CacheAffinityKeys[0] out.CacheAffinityKey = out.CacheAffinityKeys[0]
@@ -0,0 +1,17 @@
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
name: easyai-capacity-controller
namespace: easyai
rules:
- apiGroups: ["apps"]
resources: ["deployments"]
resourceNames: ["easyai-worker-ningbo", "easyai-worker-hongkong"]
verbs: ["get"]
- apiGroups: ["apps"]
resources: ["deployments/scale"]
resourceNames: ["easyai-worker-ningbo", "easyai-worker-hongkong"]
verbs: ["get", "update", "patch"]
- apiGroups: [""]
resources: ["pods"]
verbs: ["get", "list", "patch"]
+8
View File
@@ -599,12 +599,20 @@ render_and_apply_application() {
chmod 0600 "$rendered" chmod 0600 "$rendered"
kubectl --context "$context" -n "$namespace" apply \ kubectl --context "$context" -n "$namespace" apply \
-f "$repository_root/deploy/kubernetes/production/service-account-rbac.yaml" >/dev/null -f "$repository_root/deploy/kubernetes/production/service-account-rbac.yaml" >/dev/null
kubectl --context "$context" -n "$namespace" apply \
-f "$manifest_root/capacity-controller-rbac.yaml" >/dev/null
kubectl --context "$context" -n "$namespace" apply -f "$manifest_root/local-config.yaml" >/dev/null kubectl --context "$context" -n "$namespace" apply -f "$manifest_root/local-config.yaml" >/dev/null
kubectl --context "$context" -n "$namespace" apply -f "$rendered" >/dev/null kubectl --context "$context" -n "$namespace" apply -f "$rendered" >/dev/null
kubectl --context "$context" -n "$namespace" scale \
deployment/easyai-worker-hongkong --replicas=0 >/dev/null
kubectl --context "$context" -n "$namespace" rollout status \
deployment/easyai-worker-hongkong --timeout=5m >/dev/null
for workload in easyai-worker-ningbo easyai-worker-hongkong; do for workload in easyai-worker-ningbo easyai-worker-hongkong; do
kubectl --context "$context" -n "$namespace" patch deployment "$workload" \ kubectl --context "$context" -n "$namespace" patch deployment "$workload" \
--type=merge -p='{"spec":{"template":{"spec":{"affinity":null}}}}' >/dev/null --type=merge -p='{"spec":{"template":{"spec":{"affinity":null}}}}' >/dev/null
done done
kubectl --context "$context" -n "$namespace" scale \
deployment/easyai-worker-hongkong --replicas=2 >/dev/null
for workload in easyai-api-ningbo easyai-worker-ningbo; do for workload in easyai-api-ningbo easyai-worker-ningbo; do
kubectl --context "$context" -n "$namespace" set env deployment/"$workload" \ kubectl --context "$context" -n "$namespace" set env deployment/"$workload" \
+3
View File
@@ -543,6 +543,9 @@ provider_burst() {
sample_resources "$resources" "$worker_sample_stop" & sample_resources "$resources" "$worker_sample_stop" &
active_resource_sampler_pid=$! active_resource_sampler_pid=$!
current_phase=provider_burst_load current_phase=provider_burst_load
while ((10#$(date -u '+%S') > 5)); do
sleep 1
done
run_load video-provider-quota "$load_report" -requests "$requests" run_load video-provider-quota "$load_report" -requests "$requests"
sleep 2 sleep 2
touch "$worker_sample_stop" touch "$worker_sample_stop"