test(acceptance): 验证厂商三类额度与优先路由
为 provider-burst 固定三平台并发、RPM、TPM 和优先级组合,使用稳定 token 用量的本地视频负载验证同级按容量分流及低优先级溢出。\n\n报告升级为 v2,强制校验三类额度峰值精确命中上限、不越界、队列回收、三 Worker 分发以及重复提交和回调安全。\n\n验证:\n- env -u AI_GATEWAY_TEST_DATABASE_URL go test ./... -count=1\n- go vet ./...\n- bash -n scripts/acceptance/provider-burst.sh scripts/acceptance/run-local-acceptance.sh\n- shellcheck -x scripts/acceptance/provider-burst.sh scripts/acceptance/run-local-acceptance.sh
This commit is contained in:
@@ -279,7 +279,7 @@ func parseOptions() (options, error) {
|
|||||||
return options{}, errors.New("requests must be between 0 and 10000")
|
return options{}, errors.New("requests must be between 0 and 10000")
|
||||||
}
|
}
|
||||||
switch opts.profile {
|
switch opts.profile {
|
||||||
case "simulated-smoke", "simulated-all", "gemini-baseline", "gemini-multi-image", "gemini-large", "gemini-peak", "video-throughput", "video-capacity", "video-recovery",
|
case "simulated-smoke", "simulated-all", "gemini-baseline", "gemini-multi-image", "gemini-large", "gemini-peak", "video-throughput", "video-capacity", "video-provider-quota", "video-recovery",
|
||||||
"mixed-soak", "mixed-overload":
|
"mixed-soak", "mixed-overload":
|
||||||
if opts.emulatorURL == "" {
|
if opts.emulatorURL == "" {
|
||||||
return options{}, errors.New("AI_GATEWAY_ACCEPTANCE_EMULATOR_URL is required for simulated profiles")
|
return options{}, errors.New("AI_GATEWAY_ACCEPTANCE_EMULATOR_URL is required for simulated profiles")
|
||||||
@@ -371,6 +371,12 @@ func run(ctx context.Context, opts options) ([]phaseResult, error) {
|
|||||||
requests = opts.requestCount
|
requests = opts.requestCount
|
||||||
}
|
}
|
||||||
result = runVideo(ctx, client, opts, name, opts.shardRequestCount(requests), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
|
result = runVideo(ctx, client, opts, name, opts.shardRequestCount(requests), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
|
||||||
|
case "video-provider-quota":
|
||||||
|
requests := 24
|
||||||
|
if opts.requestCount > 0 {
|
||||||
|
requests = opts.requestCount
|
||||||
|
}
|
||||||
|
result = runVideo(ctx, client, opts, name, opts.shardRequestCount(requests), false, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
|
||||||
case "video-recovery":
|
case "video-recovery":
|
||||||
result = runVideo(ctx, client, opts, name, opts.shardRequestCount(96), true, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
|
result = runVideo(ctx, client, opts, name, opts.shardRequestCount(96), true, opts.emulatorFixtureURLs(), false, opts.shardIndex, opts.shardCount)
|
||||||
case "mixed-soak", "mixed-overload":
|
case "mixed-soak", "mixed-overload":
|
||||||
@@ -657,18 +663,21 @@ func runVideo(
|
|||||||
if len(combo) > imageCount {
|
if len(combo) > imageCount {
|
||||||
combo = combo[:imageCount]
|
combo = combo[:imageCount]
|
||||||
}
|
}
|
||||||
if !realUpstream && logicalIndex%4 == 0 {
|
if !realUpstream && name != "video-provider-quota" && logicalIndex%4 == 0 {
|
||||||
combo[0] = imageURLs[12+(logicalIndex/4)%4]
|
combo[0] = imageURLs[12+(logicalIndex/4)%4]
|
||||||
}
|
}
|
||||||
content := make([]any, 0, len(combo)+1)
|
content := make([]any, 0, len(combo)+1)
|
||||||
prompt := "多参考图生成连续运镜视频,保持人物、服装和场景一致"
|
prompt := "多参考图生成连续运镜视频,保持人物、服装和场景一致"
|
||||||
|
if name == "video-provider-quota" {
|
||||||
|
prompt = "acceptance-provider-quota"
|
||||||
|
}
|
||||||
if longRun {
|
if longRun {
|
||||||
prompt += " acceptance-long-recovery"
|
prompt += " acceptance-long-recovery"
|
||||||
}
|
}
|
||||||
if name == "video-capacity" {
|
if name == "video-capacity" {
|
||||||
prompt += " acceptance-capacity-ladder"
|
prompt += " acceptance-capacity-ladder"
|
||||||
}
|
}
|
||||||
if !realUpstream && logicalIndex%4 == 0 {
|
if !realUpstream && name != "video-provider-quota" && logicalIndex%4 == 0 {
|
||||||
prompt += " acceptance-force-conversion"
|
prompt += " acceptance-force-conversion"
|
||||||
}
|
}
|
||||||
content = append(content, map[string]any{"type": "text", "text": prompt})
|
content = append(content, map[string]any{"type": "text", "text": prompt})
|
||||||
@@ -716,7 +725,7 @@ func runVideo(
|
|||||||
err = withOperation("video_submit", err)
|
err = withOperation("video_submit", err)
|
||||||
mu.Lock()
|
mu.Lock()
|
||||||
latencies[index] = time.Since(requestStarted)
|
latencies[index] = time.Since(requestStarted)
|
||||||
if !realUpstream && logicalIndex%4 == 0 {
|
if !realUpstream && name != "video-provider-quota" && logicalIndex%4 == 0 {
|
||||||
forcedConversions++
|
forcedConversions++
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -78,11 +78,12 @@ type Report struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type videoTask struct {
|
type videoTask struct {
|
||||||
ID string
|
ID string
|
||||||
Model string
|
Model string
|
||||||
CreatedAt time.Time
|
CreatedAt time.Time
|
||||||
ReadyAt time.Time
|
ReadyAt time.Time
|
||||||
ImageRefs []imageReference
|
UsageTokens int
|
||||||
|
ImageRefs []imageReference
|
||||||
}
|
}
|
||||||
|
|
||||||
type imageReference struct {
|
type imageReference struct {
|
||||||
@@ -243,7 +244,7 @@ func (s *Server) submitVideo(w http.ResponseWriter, r *http.Request) {
|
|||||||
id := "acceptance-video-" + strconv.FormatUint(s.nextID.Add(1), 10)
|
id := "acceptance-video-" + strconv.FormatUint(s.nextID.Add(1), 10)
|
||||||
task := videoTask{
|
task := videoTask{
|
||||||
ID: id, Model: strings.TrimSpace(stringValue(body["model"])),
|
ID: id, Model: strings.TrimSpace(stringValue(body["model"])),
|
||||||
CreatedAt: s.now(), ReadyAt: s.now().Add(delay), ImageRefs: refs,
|
CreatedAt: s.now(), ReadyAt: s.now().Add(delay), UsageTokens: videoUsageTokens(body), ImageRefs: refs,
|
||||||
}
|
}
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
if existingID := s.videoByIdempotency[idempotencyKey]; existingID != "" {
|
if existingID := s.videoByIdempotency[idempotencyKey]; existingID != "" {
|
||||||
@@ -298,7 +299,7 @@ func (s *Server) getVideo(w http.ResponseWriter, r *http.Request) {
|
|||||||
writeJSON(w, http.StatusOK, map[string]any{
|
writeJSON(w, http.StatusOK, map[string]any{
|
||||||
"id": id, "model": task.Model, "status": status, "content": content,
|
"id": id, "model": task.Model, "status": status, "content": content,
|
||||||
"created_at": task.CreatedAt.Unix(),
|
"created_at": task.CreatedAt.Unix(),
|
||||||
"usage": map[string]any{"completion_tokens": 1, "total_tokens": 1},
|
"usage": map[string]any{"completion_tokens": task.UsageTokens, "total_tokens": task.UsageTokens},
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -674,12 +675,22 @@ func videoDelay(body map[string]any, longRun bool) time.Duration {
|
|||||||
if longRun {
|
if longRun {
|
||||||
return 2*time.Minute + time.Duration(seed%61)*time.Second
|
return 2*time.Minute + time.Duration(seed%61)*time.Second
|
||||||
}
|
}
|
||||||
|
if videoPromptContains(body, "acceptance-provider-quota") {
|
||||||
|
return 5*time.Second + time.Duration(seed%6)*time.Second
|
||||||
|
}
|
||||||
if videoPromptContains(body, "acceptance-capacity-ladder") {
|
if videoPromptContains(body, "acceptance-capacity-ladder") {
|
||||||
return 30*time.Second + time.Duration(seed%16)*time.Second
|
return 30*time.Second + time.Duration(seed%16)*time.Second
|
||||||
}
|
}
|
||||||
return 5*time.Second + time.Duration(seed%11)*time.Second
|
return 5*time.Second + time.Duration(seed%11)*time.Second
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func videoUsageTokens(body map[string]any) int {
|
||||||
|
if videoPromptContains(body, "acceptance-provider-quota") {
|
||||||
|
return 7
|
||||||
|
}
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
|
||||||
func videoPromptContains(body map[string]any, marker string) bool {
|
func videoPromptContains(body map[string]any, marker string) bool {
|
||||||
content, _ := body["content"].([]any)
|
content, _ := body["content"].([]any)
|
||||||
for _, raw := range content {
|
for _, raw := range content {
|
||||||
|
|||||||
@@ -242,6 +242,23 @@ func TestVideoCapacityDelay(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestVideoProviderQuotaDelayAndUsage(t *testing.T) {
|
||||||
|
body := map[string]any{
|
||||||
|
"seed": json.Number("5"),
|
||||||
|
"content": []any{map[string]any{
|
||||||
|
"type": "text",
|
||||||
|
"text": "acceptance-provider-quota",
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
delay := videoDelay(body, false)
|
||||||
|
if delay < 5*time.Second || delay > 10*time.Second {
|
||||||
|
t.Fatalf("video provider quota delay=%s, want 5s..10s", delay)
|
||||||
|
}
|
||||||
|
if tokens := videoUsageTokens(body); tokens != 7 {
|
||||||
|
t.Fatalf("video provider quota usage tokens=%d, want 7", tokens)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestVolcesProtocolAcceptsThreeSixAndNineReferenceImages(t *testing.T) {
|
func TestVolcesProtocolAcceptsThreeSixAndNineReferenceImages(t *testing.T) {
|
||||||
server := New(Config{Wait: func(context.Context, time.Duration) error { return nil }})
|
server := New(Config{Wait: func(context.Context, time.Duration) error { return nil }})
|
||||||
httpServer := httptest.NewServer(server.Handler())
|
httpServer := httptest.NewServer(server.Handler())
|
||||||
|
|||||||
@@ -134,3 +134,15 @@ func TestReservationsFromPolicySkipsNonPositiveLimits(t *testing.T) {
|
|||||||
t.Fatalf("expected concurrent reservation with limit 2, got %+v", reservations[0])
|
t.Fatalf("expected concurrent reservation with limit 2, got %+v", reservations[0])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestProviderQuotaPromptReservesSevenTokens(t *testing.T) {
|
||||||
|
body := map[string]any{
|
||||||
|
"content": []any{map[string]any{
|
||||||
|
"type": "text",
|
||||||
|
"text": "acceptance-provider-quota",
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
if got := estimateRequestTokens(body); got != 7 {
|
||||||
|
t.Fatalf("provider quota prompt tokens=%d, want 7", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -44,11 +44,11 @@ WITH source AS (
|
|||||||
AND COALESCE((platform.config->>'acceptanceSnapshot')::boolean, false)
|
AND COALESCE((platform.config->>'acceptanceSnapshot')::boolean, false)
|
||||||
ORDER BY platform.priority, platform.created_at
|
ORDER BY platform.priority, platform.created_at
|
||||||
LIMIT 1
|
LIMIT 1
|
||||||
), variants(platform_id, platform_key, platform_name, concurrency_limit) AS (
|
), variants(platform_id, platform_key, platform_name, concurrency_limit, rpm_limit, tpm_limit, priority) AS (
|
||||||
VALUES
|
VALUES
|
||||||
('${provider_burst_platform_ids[0]}'::uuid, 'acceptance-provider-burst-2', 'Acceptance Provider Burst C2', 2),
|
('${provider_burst_platform_ids[0]}'::uuid, 'acceptance-provider-burst-2', 'Acceptance Provider Burst C2', 2, 4, 28, 100),
|
||||||
('${provider_burst_platform_ids[1]}'::uuid, 'acceptance-provider-burst-4', 'Acceptance Provider Burst C4', 4),
|
('${provider_burst_platform_ids[1]}'::uuid, 'acceptance-provider-burst-4', 'Acceptance Provider Burst C4', 4, 8, 56, 100),
|
||||||
('${provider_burst_platform_ids[2]}'::uuid, 'acceptance-provider-burst-6', 'Acceptance Provider Burst C6', 6)
|
('${provider_burst_platform_ids[2]}'::uuid, 'acceptance-provider-burst-6', 'Acceptance Provider Burst C6', 6, 12, 84, 200)
|
||||||
)
|
)
|
||||||
INSERT INTO integration_platforms (
|
INSERT INTO integration_platforms (
|
||||||
id, provider, platform_key, name, base_url, auth_type, credentials, config,
|
id, provider, platform_key, name, base_url, auth_type, credentials, config,
|
||||||
@@ -61,12 +61,14 @@ SELECT variants.platform_id, source.provider, variants.platform_key, variants.pl
|
|||||||
source.config || jsonb_build_object(
|
source.config || jsonb_build_object(
|
||||||
'acceptanceProviderBurst', true,
|
'acceptanceProviderBurst', true,
|
||||||
'acceptanceBurstRunId', '$run_id',
|
'acceptanceBurstRunId', '$run_id',
|
||||||
'acceptanceConcurrencyLimit', variants.concurrency_limit
|
'acceptanceConcurrencyLimit', variants.concurrency_limit,
|
||||||
|
'acceptanceRPMLimit', variants.rpm_limit,
|
||||||
|
'acceptanceTPMLimit', variants.tpm_limit
|
||||||
),
|
),
|
||||||
source.visibility_scope, source.tenant_id, source.tenant_key,
|
source.visibility_scope, source.tenant_id, source.tenant_key,
|
||||||
source.default_pricing_mode, source.default_discount_factor,
|
source.default_pricing_mode, source.default_discount_factor,
|
||||||
source.pricing_rule_set_id, source.retry_policy, '{\"rules\":[]}'::jsonb,
|
source.pricing_rule_set_id, source.retry_policy, '{\"rules\":[]}'::jsonb,
|
||||||
100, NULL, 'enabled', NULL, NULL, variants.platform_name
|
variants.priority, NULL, 'enabled', NULL, NULL, variants.platform_name
|
||||||
FROM source CROSS JOIN variants
|
FROM source CROSS JOIN variants
|
||||||
ON CONFLICT (id) DO UPDATE SET
|
ON CONFLICT (id) DO UPDATE SET
|
||||||
provider = EXCLUDED.provider,
|
provider = EXCLUDED.provider,
|
||||||
@@ -102,11 +104,11 @@ WITH source AS (
|
|||||||
AND COALESCE((platform.config->>'acceptanceSnapshot')::boolean, false)
|
AND COALESCE((platform.config->>'acceptanceSnapshot')::boolean, false)
|
||||||
ORDER BY platform.priority, platform.created_at
|
ORDER BY platform.priority, platform.created_at
|
||||||
LIMIT 1
|
LIMIT 1
|
||||||
), variants(model_id, platform_id, concurrency_limit) AS (
|
), variants(model_id, platform_id, concurrency_limit, rpm_limit, tpm_limit) AS (
|
||||||
VALUES
|
VALUES
|
||||||
('${provider_burst_model_ids[0]}'::uuid, '${provider_burst_platform_ids[0]}'::uuid, 2),
|
('${provider_burst_model_ids[0]}'::uuid, '${provider_burst_platform_ids[0]}'::uuid, 2, 4, 28),
|
||||||
('${provider_burst_model_ids[1]}'::uuid, '${provider_burst_platform_ids[1]}'::uuid, 4),
|
('${provider_burst_model_ids[1]}'::uuid, '${provider_burst_platform_ids[1]}'::uuid, 4, 8, 56),
|
||||||
('${provider_burst_model_ids[2]}'::uuid, '${provider_burst_platform_ids[2]}'::uuid, 6)
|
('${provider_burst_model_ids[2]}'::uuid, '${provider_burst_platform_ids[2]}'::uuid, 6, 12, 84)
|
||||||
)
|
)
|
||||||
INSERT INTO platform_models (
|
INSERT INTO platform_models (
|
||||||
id, platform_id, base_model_id, model_name, model_alias, model_type,
|
id, platform_id, base_model_id, model_name, model_alias, model_type,
|
||||||
@@ -127,7 +129,8 @@ SELECT variants.model_id, variants.platform_id, source.base_model_id,
|
|||||||
'metric', 'concurrent', 'limit', variants.concurrency_limit,
|
'metric', 'concurrent', 'limit', variants.concurrency_limit,
|
||||||
'leaseTtlSeconds', 120
|
'leaseTtlSeconds', 120
|
||||||
),
|
),
|
||||||
jsonb_build_object('metric', 'rpm', 'limit', 600, 'windowSeconds', 60)
|
jsonb_build_object('metric', 'rpm', 'limit', variants.rpm_limit, 'windowSeconds', 60),
|
||||||
|
jsonb_build_object('metric', 'tpm_total', 'limit', variants.tpm_limit, 'windowSeconds', 60)
|
||||||
)),
|
)),
|
||||||
'override', NULL, '{}'::jsonb, source.provider_model_name, NULL, true
|
'override', NULL, '{}'::jsonb, source.provider_model_name, NULL, true
|
||||||
FROM source CROSS JOIN variants
|
FROM source CROSS JOIN variants
|
||||||
@@ -219,13 +222,18 @@ DO UPDATE SET status = 'active', metadata = EXCLUDED.metadata,
|
|||||||
SELECT jsonb_build_object(
|
SELECT jsonb_build_object(
|
||||||
'platforms', count(*),
|
'platforms', count(*),
|
||||||
'limits', jsonb_agg((platform.config->>'acceptanceConcurrencyLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int),
|
'limits', jsonb_agg((platform.config->>'acceptanceConcurrencyLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int),
|
||||||
|
'rpmLimits', jsonb_agg((platform.config->>'acceptanceRPMLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int),
|
||||||
|
'tpmLimits', jsonb_agg((platform.config->>'acceptanceTPMLimit')::int ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int),
|
||||||
|
'priorities', jsonb_agg(platform.priority ORDER BY (platform.config->>'acceptanceConcurrencyLimit')::int),
|
||||||
'enabled', count(*) FILTER (WHERE platform.status='enabled' AND model.enabled)
|
'enabled', count(*) FILTER (WHERE platform.status='enabled' AND model.enabled)
|
||||||
)
|
)
|
||||||
FROM integration_platforms platform
|
FROM integration_platforms platform
|
||||||
JOIN platform_models model ON model.platform_id = platform.id
|
JOIN platform_models model ON model.platform_id = platform.id
|
||||||
WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false)
|
WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false)
|
||||||
AND platform.config->>'acceptanceBurstRunId' = '$run_id';")
|
AND platform.config->>'acceptanceBurstRunId' = '$run_id';")
|
||||||
jq -e '.platforms == 3 and .enabled == 3 and .limits == [2,4,6]' <<<"$configured" >/dev/null
|
jq -e '.platforms == 3 and .enabled == 3 and .limits == [2,4,6] and
|
||||||
|
.rpmLimits == [4,8,12] and .tpmLimits == [28,56,84] and .priorities == [100,100,200]' \
|
||||||
|
<<<"$configured" >/dev/null
|
||||||
}
|
}
|
||||||
|
|
||||||
sample_provider_burst() {
|
sample_provider_burst() {
|
||||||
@@ -235,7 +243,10 @@ sample_provider_burst() {
|
|||||||
database_query "
|
database_query "
|
||||||
WITH configured AS (
|
WITH configured AS (
|
||||||
SELECT platform.id platform_id, platform.platform_key, model.id platform_model_id,
|
SELECT platform.id platform_id, platform.platform_key, model.id platform_model_id,
|
||||||
(platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit
|
platform.priority,
|
||||||
|
(platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit,
|
||||||
|
(platform.config->>'acceptanceRPMLimit')::int rpm_limit,
|
||||||
|
(platform.config->>'acceptanceTPMLimit')::int tpm_limit
|
||||||
FROM integration_platforms platform
|
FROM integration_platforms platform
|
||||||
JOIN platform_models model ON model.platform_id = platform.id
|
JOIN platform_models model ON model.platform_id = platform.id
|
||||||
WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false)
|
WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false)
|
||||||
@@ -302,7 +313,10 @@ build_provider_burst_report() {
|
|||||||
platforms=$(database_query "
|
platforms=$(database_query "
|
||||||
WITH configured AS (
|
WITH configured AS (
|
||||||
SELECT platform.id platform_id, platform.platform_key, model.id platform_model_id,
|
SELECT platform.id platform_id, platform.platform_key, model.id platform_model_id,
|
||||||
(platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit
|
platform.priority,
|
||||||
|
(platform.config->>'acceptanceConcurrencyLimit')::int concurrency_limit,
|
||||||
|
(platform.config->>'acceptanceRPMLimit')::int rpm_limit,
|
||||||
|
(platform.config->>'acceptanceTPMLimit')::int tpm_limit
|
||||||
FROM integration_platforms platform
|
FROM integration_platforms platform
|
||||||
JOIN platform_models model ON model.platform_id=platform.id
|
JOIN platform_models model ON model.platform_id=platform.id
|
||||||
WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false)
|
WHERE COALESCE((platform.config->>'acceptanceProviderBurst')::boolean, false)
|
||||||
@@ -332,18 +346,34 @@ WITH configured AS (
|
|||||||
FROM gateway_task_attempts attempt
|
FROM gateway_task_attempts attempt
|
||||||
WHERE attempt.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid)
|
WHERE attempt.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid)
|
||||||
GROUP BY attempt.platform_model_id
|
GROUP BY attempt.platform_model_id
|
||||||
|
), rate_peaks AS (
|
||||||
|
SELECT split_part(counter.scope_key, ':', 3)::uuid platform_model_id,
|
||||||
|
max(counter.limit_value) FILTER (WHERE counter.metric='rpm') rpm_limit,
|
||||||
|
max(counter.used_value + counter.reserved_value) FILTER (WHERE counter.metric='rpm') rpm_peak,
|
||||||
|
max(counter.limit_value) FILTER (WHERE counter.metric='tpm_total') tpm_limit,
|
||||||
|
max(counter.used_value + counter.reserved_value) FILTER (WHERE counter.metric='tpm_total') tpm_peak
|
||||||
|
FROM gateway_rate_limit_counters counter
|
||||||
|
WHERE counter.scope_type='platform_model'
|
||||||
|
AND counter.scope_key LIKE 'acceptance:$run_id:%'
|
||||||
|
GROUP BY split_part(counter.scope_key, ':', 3)::uuid
|
||||||
)
|
)
|
||||||
SELECT COALESCE(jsonb_agg(jsonb_build_object(
|
SELECT COALESCE(jsonb_agg(jsonb_build_object(
|
||||||
'platformId', configured.platform_id,
|
'platformId', configured.platform_id,
|
||||||
'platformModelId', configured.platform_model_id,
|
'platformModelId', configured.platform_model_id,
|
||||||
'platformKey', configured.platform_key,
|
'platformKey', configured.platform_key,
|
||||||
|
'priority', configured.priority,
|
||||||
'limit', configured.concurrency_limit,
|
'limit', configured.concurrency_limit,
|
||||||
'peak', COALESCE(peaks.peak,0),
|
'peak', COALESCE(peaks.peak,0),
|
||||||
|
'rpmLimit', configured.rpm_limit,
|
||||||
|
'rpmPeak', COALESCE(rate_peaks.rpm_peak,0),
|
||||||
|
'tpmLimit', configured.tpm_limit,
|
||||||
|
'tpmPeak', COALESCE(rate_peaks.tpm_peak,0),
|
||||||
'attempts', COALESCE(attempts.attempts,0)
|
'attempts', COALESCE(attempts.attempts,0)
|
||||||
) ORDER BY configured.concurrency_limit), '[]'::jsonb)
|
) ORDER BY configured.concurrency_limit), '[]'::jsonb)
|
||||||
FROM configured
|
FROM configured
|
||||||
LEFT JOIN peaks ON peaks.platform_model_id=configured.platform_model_id
|
LEFT JOIN peaks ON peaks.platform_model_id=configured.platform_model_id
|
||||||
LEFT JOIN attempts ON attempts.platform_model_id=configured.platform_model_id;")
|
LEFT JOIN attempts ON attempts.platform_model_id=configured.platform_model_id
|
||||||
|
LEFT JOIN rate_peaks ON rate_peaks.platform_model_id=configured.platform_model_id;")
|
||||||
queue=$(jq -s '{
|
queue=$(jq -s '{
|
||||||
samples:length,
|
samples:length,
|
||||||
maxQueuedTasks:(map(.queuedTasks)|max),
|
maxQueuedTasks:(map(.queuedTasks)|max),
|
||||||
@@ -437,7 +467,16 @@ SELECT jsonb_build_object(
|
|||||||
FROM gateway_task_callback_outbox callback
|
FROM gateway_task_callback_outbox callback
|
||||||
WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid);")
|
WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id='$run_id'::uuid);")
|
||||||
resource_summary=$(provider_burst_resource_summary "$resources")
|
resource_summary=$(provider_burst_resource_summary "$resources")
|
||||||
if ! jq -e 'length==3 and all(.[]; .peak == .limit and .attempts > 0)' <<<"$platforms" >/dev/null; then passed=false; fi
|
if ! jq -e 'length==3 and
|
||||||
|
[.[].priority] == [100,100,200] and
|
||||||
|
[.[].attempts] == [4,8,12] and
|
||||||
|
all(.[]; .peak == .limit and .rpmPeak == .rpmLimit and .tpmPeak == .tpmLimit)' \
|
||||||
|
<<<"$platforms" >/dev/null; then passed=false; fi
|
||||||
|
if ! jq -e '.[0].priority==.[1].priority and
|
||||||
|
.[0].attempts*2==.[1].attempts and
|
||||||
|
.[2].priority>.[1].priority and
|
||||||
|
.[0].peak==.[0].limit and .[1].peak==.[1].limit and .[2].attempts>0' \
|
||||||
|
<<<"$platforms" >/dev/null; then passed=false; fi
|
||||||
if ! jq -e '.queueObserved and .admissionWaitObserved and .drainedAtEnd and .maxQueuedTasks > 0 and .maxWaitingAdmissions > 0' <<<"$queue" >/dev/null; then passed=false; fi
|
if ! jq -e '.queueObserved and .admissionWaitObserved and .drainedAtEnd and .maxQueuedTasks > 0 and .maxWaitingAdmissions > 0' <<<"$queue" >/dev/null; then passed=false; fi
|
||||||
if ! jq -e --argjson requests "$requests" 'length==3 and all(.[]; .tasks>0 and .workerInstanceId!="unclaimed") and ([.[].tasks]|add)==$requests' <<<"$distribution" >/dev/null; then passed=false; fi
|
if ! jq -e --argjson requests "$requests" 'length==3 and all(.[]; .tasks>0 and .workerInstanceId!="unclaimed") and ([.[].tasks]|add)==$requests' <<<"$distribution" >/dev/null; then passed=false; fi
|
||||||
if ! jq -e 'length==3 and all(.[]; .peakActiveTasks>0 and (.pressureStates|index("critical")|not))' <<<"$worker_peaks" >/dev/null; then passed=false; fi
|
if ! jq -e 'length==3 and all(.[]; .peakActiveTasks>0 and (.pressureStates|index("critical")|not))' <<<"$worker_peaks" >/dev/null; then passed=false; fi
|
||||||
@@ -458,11 +497,20 @@ WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id=
|
|||||||
--argjson resources "$resource_summary" --argjson leaks "$leaks" \
|
--argjson resources "$resource_summary" --argjson leaks "$leaks" \
|
||||||
--argjson duplicates "$duplicates" --argjson callbacks "$callbacks" \
|
--argjson duplicates "$duplicates" --argjson callbacks "$callbacks" \
|
||||||
'{
|
'{
|
||||||
schemaVersion:"acceptance-provider-burst-report/v1",
|
schemaVersion:"acceptance-provider-burst-report/v2",
|
||||||
runId:$runId,profile:"provider-burst",model:$model,
|
runId:$runId,profile:"provider-burst",model:$model,
|
||||||
startedAt:(now|todateiso8601),finishedAt:(now|todateiso8601),
|
startedAt:(now|todateiso8601),finishedAt:(now|todateiso8601),
|
||||||
passed:$passed,secretSafe:true,requests:$requests,phases:$phases,
|
passed:$passed,secretSafe:true,requests:$requests,phases:$phases,
|
||||||
providerRouting:{platforms:$platforms,totalConfiguredConcurrency:([$platforms[].limit]|add)},
|
providerRouting:{
|
||||||
|
platforms:$platforms,
|
||||||
|
totalConfiguredConcurrency:([$platforms[].limit]|add),
|
||||||
|
samePriorityLoadBalanced:($platforms[0].priority==$platforms[1].priority and
|
||||||
|
$platforms[0].attempts*2==$platforms[1].attempts),
|
||||||
|
lowerPrioritySpilloverObserved:($platforms[2].priority>$platforms[1].priority and
|
||||||
|
$platforms[0].peak==$platforms[0].limit and
|
||||||
|
$platforms[1].peak==$platforms[1].limit and
|
||||||
|
$platforms[2].attempts>0)
|
||||||
|
},
|
||||||
queue:$queue,
|
queue:$queue,
|
||||||
cluster:{workers:$workers,workerPeaks:$workerPeaks,taskDistribution:$distribution,
|
cluster:{workers:$workers,workerPeaks:$workerPeaks,taskDistribution:$distribution,
|
||||||
rateLimits:$rateLimits,concurrencyPeaks:$concurrencyPeaks,
|
rateLimits:$rateLimits,concurrencyPeaks:$concurrencyPeaks,
|
||||||
@@ -473,14 +521,14 @@ WHERE callback.task_id IN (SELECT id FROM gateway_tasks WHERE acceptance_run_id=
|
|||||||
}
|
}
|
||||||
|
|
||||||
provider_burst() {
|
provider_burst() {
|
||||||
local requests=${AI_GATEWAY_LOCAL_PROVIDER_BURST_REQUESTS:-48}
|
local requests=24
|
||||||
local load_report="$report_root/provider-burst-load.json"
|
local load_report="$report_root/provider-burst-load.json"
|
||||||
local samples="$report_root/provider-burst-samples.ndjson"
|
local samples="$report_root/provider-burst-samples.ndjson"
|
||||||
local worker_samples="$report_root/provider-burst-worker-samples.ndjson"
|
local worker_samples="$report_root/provider-burst-worker-samples.ndjson"
|
||||||
local resources="$report_root/provider-burst-resources.csv"
|
local resources="$report_root/provider-burst-resources.csv"
|
||||||
local report="$report_root/provider-burst.json"
|
local report="$report_root/provider-burst.json"
|
||||||
worker_sample_stop="$report_root/provider-burst-samples.stop"
|
worker_sample_stop="$report_root/provider-burst-samples.stop"
|
||||||
[[ $requests =~ ^[0-9]+$ ]] && ((requests >= 24 && requests <= 256))
|
[[ $requests == 24 ]]
|
||||||
[[ ! -e $load_report && ! -e $samples && ! -e $worker_samples && ! -e $resources && ! -e $report && ! -e $worker_sample_stop ]] ||
|
[[ ! -e $load_report && ! -e $samples && ! -e $worker_samples && ! -e $resources && ! -e $report && ! -e $worker_sample_stop ]] ||
|
||||||
fail_gate acceptance_report_exists "provider burst evidence already exists"
|
fail_gate acceptance_report_exists "provider burst evidence already exists"
|
||||||
current_phase=provider_burst_platforms
|
current_phase=provider_burst_platforms
|
||||||
@@ -495,7 +543,7 @@ 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
|
||||||
run_load video-throughput "$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"
|
||||||
wait "$active_provider_sampler_pid"
|
wait "$active_provider_sampler_pid"
|
||||||
|
|||||||
Reference in New Issue
Block a user