zizi-al/sub2api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(scheduler): add sticky health escape

Browse Source
This commit is contained in:
wucm667 2026-05-29 10:25:26 +08:00
parent 6010c3cca9
commit 415d08f255
5 changed files with 399 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
backend/internal/config/config.go
View File

@ -716,6 +716,8 @@ type GatewayConfig struct {
OpenAIPassthroughAllowTimeoutHeaders bool `mapstructure:"openai_passthrough_allow_timeout_headers"`
// OpenAIWS: OpenAI Responses WebSocket 配置（默认开启，可按需回滚到 HTTP）
OpenAIWS GatewayOpenAIWSConfig `mapstructure:"openai_ws"`
// OpenAIScheduler: OpenAI 高级调度器粘性逃逸配置
OpenAIScheduler GatewayOpenAISchedulerConfig `mapstructure:"openai_scheduler"`
// OpenAIHTTP2: OpenAI HTTP 上游协议策略（默认启用 HTTP/2，可按代理能力回退 HTTP/1.1）
OpenAIHTTP2 GatewayOpenAIHTTP2Config `mapstructure:"openai_http2"`
// ImageConcurrency: 图片生成独立并发限制配置（默认关闭）
@ -948,6 +950,16 @@ type GatewayOpenAIWSSchedulerScoreWeights struct {
TTFT float64 `mapstructure:"ttft"`
}
// GatewayOpenAISchedulerConfig OpenAI 高级调度器配置。
type GatewayOpenAISchedulerConfig struct {
// StickyEscapeEnabled: 是否允许 session_hash sticky 在账号健康度劣化时临时逃逸
StickyEscapeEnabled bool `mapstructure:"sticky_escape_enabled"`
// StickyEscapeTTFTMs: TTFT EWMA 超过该阈值时跳过 sticky
StickyEscapeTTFTMs int `mapstructure:"sticky_escape_ttft_ms"`
// StickyEscapeErrorRate: 错误率 EWMA 超过该阈值时跳过 sticky
StickyEscapeErrorRate float64 `mapstructure:"sticky_escape_error_rate"`
}
// GatewayUsageRecordConfig 使用量记录异步队列配置
type GatewayUsageRecordConfig struct {
// WorkerCount: worker 初始数量（自动扩缩容开启时作为初始并发上限）
@ -1369,6 +1381,15 @@ func load(allowMissingJWTSecret bool) (*Config, error) {
if err := viper.Unmarshal(&cfg); err != nil {
return nil, fmt.Errorf("unmarshal config error: %w", err)
}
if cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs == 0 {
cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000
}
if cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate == 0 {
cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 0.5
}
if !cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled && !viper.IsSet("gateway.openai_scheduler.sticky_escape_enabled") {
cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled = true
}
cfg.RunMode = NormalizeRunMode(cfg.RunMode)
cfg.Server.Mode = strings.ToLower(strings.TrimSpace(cfg.Server.Mode))
@ -2603,6 +2624,12 @@ func (c *Config) Validate() error {
if weightSum <= 0 {
return fmt.Errorf("gateway.openai_ws.scheduler_score_weights must not all be zero")
}
if c.Gateway.OpenAIScheduler.StickyEscapeTTFTMs <= 0 {
return fmt.Errorf("gateway.openai_scheduler.sticky_escape_ttft_ms must be positive")
}
if c.Gateway.OpenAIScheduler.StickyEscapeErrorRate < 0 || c.Gateway.OpenAIScheduler.StickyEscapeErrorRate > 1 {
return fmt.Errorf("gateway.openai_scheduler.sticky_escape_error_rate must be between 0 and 1")
}
if c.Gateway.MaxLineSize < 0 {
return fmt.Errorf("gateway.max_line_size must be non-negative")
}

24
backend/internal/config/config_test.go
View File

@ -110,6 +110,15 @@ func TestLoadDefaultOpenAIWSConfig(t *testing.T) {
if cfg.Gateway.OpenAIWS.StickySessionTTLSeconds != 3600 {
t.Fatalf("Gateway.OpenAIWS.StickySessionTTLSeconds = %d, want 3600", cfg.Gateway.OpenAIWS.StickySessionTTLSeconds)
}
if !cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled {
t.Fatalf("Gateway.OpenAIScheduler.StickyEscapeEnabled = false, want true")
}
if cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs != 15000 {
t.Fatalf("Gateway.OpenAIScheduler.StickyEscapeTTFTMs = %d, want 15000", cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs)
}
if cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate != 0.5 {
t.Fatalf("Gateway.OpenAIScheduler.StickyEscapeErrorRate = %v, want 0.5", cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate)
}
if !cfg.Gateway.OpenAIWS.SessionHashReadOldFallback {
t.Fatalf("Gateway.OpenAIWS.SessionHashReadOldFallback = false, want true")
}
@ -1705,6 +1714,21 @@ func TestValidateConfig_OpenAIWSRules(t *testing.T) {
},
wantErr: "gateway.openai_ws.scheduler_score_weights must not all be zero",
},
{
name: "sticky_escape_ttft_ms 必须为正数",
mutate: func(c *Config) { c.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 0 },
wantErr: "gateway.openai_scheduler.sticky_escape_ttft_ms",
},
{
name: "sticky_escape_error_rate 不能小于 0",
mutate: func(c *Config) { c.Gateway.OpenAIScheduler.StickyEscapeErrorRate = -0.1 },
wantErr: "gateway.openai_scheduler.sticky_escape_error_rate",
},
{
name: "sticky_escape_error_rate 不能大于 1",
mutate: func(c *Config) { c.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 1.1 },
wantErr: "gateway.openai_scheduler.sticky_escape_error_rate",
},
}
for _, tc := range cases {

105
backend/internal/service/openai_account_scheduler.go
View File

@ -41,6 +41,7 @@ type OpenAIAccountScheduleRequest struct {
GroupID *int64
SessionHash string
StickyAccountID int64
PreserveStickyBinding bool
PreviousResponseID string
RequestedModel string
RequiredTransport OpenAIUpstreamTransport
@ -241,6 +242,12 @@ type defaultOpenAIAccountScheduler struct {
stats *openAIAccountRuntimeStats
}
type openAIStickyEscapeConfig struct {
enabled bool
ttftMs float64
errorRate float64
}
func newDefaultOpenAIAccountScheduler(service *OpenAIGatewayService, stats *openAIAccountRuntimeStats) OpenAIAccountScheduler {
if stats == nil {
stats = newOpenAIAccountRuntimeStats()
@ -296,7 +303,7 @@ func (s *defaultOpenAIAccountScheduler) Select(
}
}
selection, err := s.selectBySessionHash(ctx, req)
selection, escapedSticky, err := s.selectBySessionHash(ctx, req)
if err != nil {
return nil, decision, err
}
@ -307,6 +314,9 @@ func (s *defaultOpenAIAccountScheduler) Select(
decision.SelectedAccountType = selection.Account.Type
return selection, decision, nil
}
if escapedSticky {
req.PreserveStickyBinding = true
}
selection, candidateCount, topK, loadSkew, err := s.selectByLoadBalance(ctx, req)
decision.Layer = openAIAccountScheduleLayerLoadBalance
@ -326,10 +336,10 @@ func (s *defaultOpenAIAccountScheduler) Select(
func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
ctx context.Context,
req OpenAIAccountScheduleRequest,
) (*AccountSelectionResult, error) {
) (*AccountSelectionResult, bool, error) {
sessionHash := strings.TrimSpace(req.SessionHash)
if sessionHash == "" || s == nil || s.service == nil || s.service.cache == nil {
return nil, nil
return nil, false, nil
}
accountID := req.StickyAccountID
@ -337,38 +347,48 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
var err error
accountID, err = s.service.getStickySessionAccountID(ctx, req.GroupID, sessionHash)
if err != nil || accountID <= 0 {
return nil, nil
return nil, false, nil
}
}
if accountID <= 0 {
return nil, nil
return nil, false, nil
}
if req.ExcludedIDs != nil {
if _, excluded := req.ExcludedIDs[accountID]; excluded {
return nil, nil
return nil, false, nil
}
}
account, err := s.service.getSchedulableAccount(ctx, accountID)
if err != nil || account == nil {
_ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash)
return nil, nil
return nil, false, nil
}
if shouldClearStickySession(account, req.RequestedModel) || !account.IsOpenAI() || !account.IsSchedulable() {
_ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash)
return nil, nil
return nil, false, nil
}
if !s.isAccountRequestCompatible(ctx, account, req) {
return nil, nil
return nil, false, nil
}
if !s.isAccountTransportCompatible(account, req.RequiredTransport) {
_ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash)
return nil, nil
return nil, false, nil
}
account = s.service.recheckSelectedOpenAIAccountFromDB(ctx, account, req.RequestedModel, req.RequireCompact, req.RequiredCapability)
if account == nil || !s.isAccountTransportCompatible(account, req.RequiredTransport) {
_ = s.service.deleteStickySessionAccountID(ctx, req.GroupID, sessionHash)
return nil, nil
return nil, false, nil
}
escapeCfg := s.service.openAIStickyEscapeConfig()
if reason, errorRate, ttft, shouldEscape := s.shouldEscapeStickyAccount(accountID, escapeCfg); shouldEscape {
slog.Info("sticky_escape_triggered",
"account_id", accountID,
"reason", reason,
"error_rate", errorRate,
"ttft", ttft,
)
return nil, true, nil
}
result, acquireErr := s.service.tryAcquireAccountSlot(ctx, accountID, account.Concurrency)
@ -378,12 +398,22 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
Account: account,
Acquired: true,
ReleaseFunc: result.ReleaseFunc,
}, nil
}, false, nil
}
cfg := s.service.schedulingConfig()
// WaitPlan.MaxConcurrency 使用 Concurrency（非 EffectiveLoadFactor），因为 WaitPlan 控制的是 Redis 实际并发槽位等待。
if s.service.concurrencyService != nil {
if escapeCfg.enabled && acquireErr == nil && result != nil && !result.Acquired {
errorRate, ttft, _ := s.stats.snapshot(accountID)
slog.Info("sticky_escape_triggered",
"account_id", accountID,
"reason", "concurrency_full",
"error_rate", errorRate,
"ttft", ttft,
)
return nil, true, nil
}
return &AccountSelectionResult{
Account: account,
WaitPlan: &AccountWaitPlan{
@ -392,9 +422,23 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
Timeout: cfg.StickySessionWaitTimeout,
MaxWaiting: cfg.StickySessionMaxWaiting,
},
}, nil
}, false, nil
}
return nil, nil
return nil, false, nil
}
func (s *defaultOpenAIAccountScheduler) shouldEscapeStickyAccount(accountID int64, cfg openAIStickyEscapeConfig) (reason string, errorRate float64, ttft float64, shouldEscape bool) {
if !cfg.enabled || s == nil || s.stats == nil || accountID <= 0 {
return "", 0, 0, false
}
errorRate, ttft, hasTTFT := s.stats.snapshot(accountID)
if hasTTFT && ttft > cfg.ttftMs {
return "ttft", errorRate, ttft, true
}
if errorRate > cfg.errorRate {
return "error_rate", errorRate, ttft, true
}
return "", errorRate, ttft, false
}
type openAIAccountCandidateScore struct {
@ -810,7 +854,7 @@ func (s *defaultOpenAIAccountScheduler) tryAcquireOpenAISelectionOrder(
return nil, compactBlocked, acquireErr
}
if result != nil && result.Acquired {
if req.SessionHash != "" {
if req.SessionHash != "" && !req.PreserveStickyBinding {
_ = s.service.BindStickySession(ctx, req.GroupID, req.SessionHash, fresh.ID)
}
return &AccountSelectionResult{
@ -1305,6 +1349,37 @@ func (s *OpenAIGatewayService) openAIWSLBTopK() int {
return 7
}
func (s *OpenAIGatewayService) openAIStickyEscapeConfig() openAIStickyEscapeConfig {
if s != nil && s.cfg != nil {
cfg := s.cfg.Gateway.OpenAIScheduler
enabled := cfg.StickyEscapeEnabled
if !enabled && cfg.StickyEscapeTTFTMs == 0 && cfg.StickyEscapeErrorRate == 0 {
enabled = true
}
ttftMs := float64(cfg.StickyEscapeTTFTMs)
if ttftMs <= 0 {
ttftMs = 15000
}
errorRate := cfg.StickyEscapeErrorRate
if errorRate < 0 || errorRate > 1 {
errorRate = 0.5
}
if errorRate == 0 && cfg.StickyEscapeTTFTMs == 0 && cfg.StickyEscapeErrorRate == 0 {
errorRate = 0.5
}
return openAIStickyEscapeConfig{
enabled: enabled,
ttftMs: ttftMs,
errorRate: errorRate,
}
}
return openAIStickyEscapeConfig{
enabled: true,
ttftMs: 15000,
errorRate: 0.5,
}
}
func (s *OpenAIGatewayService) openAIWSSchedulerWeights() GatewayOpenAIWSSchedulerScoreWeightsView {
if s != nil && s.cfg != nil {
return GatewayOpenAIWSSchedulerScoreWeightsView{

250
backend/internal/service/openai_account_scheduler_test.go
View File

@ -948,6 +948,9 @@ func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyBusyKeepsS
cfg := &config.Config{}
cfg.Gateway.Scheduling.StickySessionMaxWaiting = 2
cfg.Gateway.Scheduling.StickySessionWaitTimeout = 45 * time.Second
cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled = false
cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000
cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 0.5
cfg.Gateway.OpenAIWS.Enabled = true
cfg.Gateway.OpenAIWS.APIKeyEnabled = true
cfg.Gateway.OpenAIWS.OAuthEnabled = true
@ -996,6 +999,253 @@ func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyBusyKeepsS
require.True(t, decision.StickySessionHit)
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyEscapeByTTFT(t *testing.T) {
ctx := context.Background()
groupID := int64(10101)
accounts := []Account{
{
ID: 21101,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Priority: 0,
},
{
ID: 21102,
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Status: StatusActive,
Schedulable: true,
Concurrency: 1,
Priority: 1,
},
}
cache := &schedulerTestGatewayCache{sessionBindings: map[string]int64{"openai:session_hash_sticky_ttft": 21101}}
cfg := &config.Config{}
cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled = true
cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000
cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 0.5
concurrencyCache := schedulerTestConcurrencyCache{acquireResults: map[int64]bool{21102: true}}
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: cache,
cfg: cfg,
rateLimitService: newOpenAIAdvancedSchedulerRateLimitService("true"),
concurrencyService: NewConcurrencyService(concurrencyCache),
openaiAccountStats: newOpenAIAccountRuntimeStats(),
}
fastTTFT := 14999
svc.openaiAccountStats.report(21101, true, &fastTTFT)
stableTTFT := 14999
svc.openaiAccountStats.report(21101, true, &stableTTFT)
selection, decision, err := svc.SelectAccountWithScheduler(ctx, &groupID, "", "session_hash_sticky_ttft", "gpt-5.1", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(21101), selection.Account.ID)
require.Equal(t, openAIAccountScheduleLayerSessionSticky, decision.Layer)
require.True(t, decision.StickySessionHit)
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
slowTTFT := 20000
for i := 0; i < 3; i++ {
svc.openaiAccountStats.report(21101, true, &slowTTFT)
}
selection, decision, err = svc.SelectAccountWithScheduler(ctx, &groupID, "", "session_hash_sticky_ttft", "gpt-5.1", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(21102), selection.Account.ID)
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
require.False(t, decision.StickySessionHit)
require.Equal(t, int64(21101), cache.sessionBindings["openai:session_hash_sticky_ttft"])
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyEscapeByErrorRate(t *testing.T) {
ctx := context.Background()
groupID := int64(10102)
accounts := []Account{
{ID: 21201, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0},
{ID: 21202, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 1},
}
cache := &schedulerTestGatewayCache{sessionBindings: map[string]int64{"openai:session_hash_sticky_error_rate": 21201}}
cfg := &config.Config{}
cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled = true
cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000
cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 0.5
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: cache,
cfg: cfg,
rateLimitService: newOpenAIAdvancedSchedulerRateLimitService("true"),
concurrencyService: NewConcurrencyService(schedulerTestConcurrencyCache{acquireResults: map[int64]bool{21202: true}}),
openaiAccountStats: newOpenAIAccountRuntimeStats(),
}
for i := 0; i < 3; i++ {
svc.openaiAccountStats.report(21201, false, nil)
}
selection, decision, err := svc.SelectAccountWithScheduler(ctx, &groupID, "", "session_hash_sticky_error_rate", "gpt-5.1", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(21201), selection.Account.ID)
require.Equal(t, openAIAccountScheduleLayerSessionSticky, decision.Layer)
require.True(t, decision.StickySessionHit)
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
for i := 0; i < 2; i++ {
svc.openaiAccountStats.report(21201, false, nil)
}
selection, decision, err = svc.SelectAccountWithScheduler(ctx, &groupID, "", "session_hash_sticky_error_rate", "gpt-5.1", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(21202), selection.Account.ID)
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
require.False(t, decision.StickySessionHit)
require.Equal(t, int64(21201), cache.sessionBindings["openai:session_hash_sticky_error_rate"])
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyBusyEscapes(t *testing.T) {
ctx := context.Background()
groupID := int64(10103)
accounts := []Account{
{ID: 21301, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0},
{ID: 21302, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 1},
}
cache := &schedulerTestGatewayCache{sessionBindings: map[string]int64{"openai:session_hash_sticky_busy_escape": 21301}}
cfg := &config.Config{}
cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled = true
cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000
cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 0.5
cfg.Gateway.Scheduling.StickySessionMaxWaiting = 2
cfg.Gateway.Scheduling.StickySessionWaitTimeout = 45 * time.Second
concurrencyCache := schedulerTestConcurrencyCache{
acquireResults: map[int64]bool{21301: false, 21302: true},
waitCounts: map[int64]int{21301: 999},
loadMap: map[int64]*AccountLoadInfo{
21301: {AccountID: 21301, LoadRate: 95, WaitingCount: 9},
21302: {AccountID: 21302, LoadRate: 1, WaitingCount: 0},
},
}
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: cache,
cfg: cfg,
rateLimitService: newOpenAIAdvancedSchedulerRateLimitService("true"),
concurrencyService: NewConcurrencyService(concurrencyCache),
}
selection, decision, err := svc.SelectAccountWithScheduler(ctx, &groupID, "", "session_hash_sticky_busy_escape", "gpt-5.1", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(21302), selection.Account.ID)
require.Nil(t, selection.WaitPlan)
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
require.False(t, decision.StickySessionHit)
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionStickyEscapeDisabledKeepsLegacyBehavior(t *testing.T) {
ctx := context.Background()
groupID := int64(10104)
accounts := []Account{
{ID: 21401, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0},
{ID: 21402, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 1},
}
cache := &schedulerTestGatewayCache{sessionBindings: map[string]int64{"openai:session_hash_sticky_disabled": 21401}}
cfg := &config.Config{}
cfg.Gateway.OpenAIScheduler.StickyEscapeEnabled = false
cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000
cfg.Gateway.OpenAIScheduler.StickyEscapeErrorRate = 0.5
cfg.Gateway.Scheduling.StickySessionMaxWaiting = 2
cfg.Gateway.Scheduling.StickySessionWaitTimeout = 45 * time.Second
concurrencyCache := schedulerTestConcurrencyCache{
acquireResults: map[int64]bool{21401: false, 21402: true},
waitCounts: map[int64]int{21401: 999},
}
svc := &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: cache,
cfg: cfg,
rateLimitService: newOpenAIAdvancedSchedulerRateLimitService("true"),
concurrencyService: NewConcurrencyService(concurrencyCache),
openaiAccountStats: newOpenAIAccountRuntimeStats(),
}
slowTTFT := 20000
svc.openaiAccountStats.report(21401, true, &slowTTFT)
for i := 0; i < 5; i++ {
svc.openaiAccountStats.report(21401, false, nil)
}
selection, decision, err := svc.SelectAccountWithScheduler(ctx, &groupID, "", "session_hash_sticky_disabled", "gpt-5.1", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
require.Equal(t, int64(21401), selection.Account.ID)
require.NotNil(t, selection.WaitPlan)
require.Equal(t, int64(21401), selection.WaitPlan.AccountID)
require.Equal(t, openAIAccountScheduleLayerSessionSticky, decision.Layer)
require.True(t, decision.StickySessionHit)
}
func TestDefaultOpenAIAccountScheduler_ShouldEscapeStickyAccount_ThresholdBoundary(t *testing.T) {
stats := newOpenAIAccountRuntimeStats()
accountID := int64(21501)
ttft := 15000
stats.report(accountID, true, &ttft)
stats.report(accountID, false, nil)
stats.report(accountID, true, nil)
scheduler := &defaultOpenAIAccountScheduler{stats: stats}
reason, errorRate, observedTTFT, shouldEscape := scheduler.shouldEscapeStickyAccount(accountID, openAIStickyEscapeConfig{
enabled: true,
ttftMs: 15000,
errorRate: 0.5,
})
require.False(t, shouldEscape)
require.Empty(t, reason)
require.InDelta(t, 0.16, errorRate, 1e-9)
require.InDelta(t, 15000, observedTTFT, 1e-9)
for i := 0; i < 4; i++ {
stats.report(accountID, false, nil)
}
reason, errorRate, _, shouldEscape = scheduler.shouldEscapeStickyAccount(accountID, openAIStickyEscapeConfig{
enabled: true,
ttftMs: 15000,
errorRate: 1,
})
require.False(t, shouldEscape)
require.Empty(t, reason)
reason, errorRate, observedTTFT, shouldEscape = scheduler.shouldEscapeStickyAccount(accountID, openAIStickyEscapeConfig{
enabled: true,
ttftMs: 15000,
errorRate: errorRate,
})
require.False(t, shouldEscape)
require.Empty(t, reason)
require.InDelta(t, 0.655936, errorRate, 1e-9)
require.InDelta(t, 15000, observedTTFT, 1e-9)
}
func TestOpenAIGatewayService_SelectAccountWithScheduler_SessionSticky_ForceHTTP(t *testing.T) {
ctx := context.Background()
groupID := int64(1010)

8
deploy/config.example.yaml
View File

@ -320,6 +320,14 @@ gateway:
queue: 0.7
error_rate: 0.8
ttft: 0.5
# OpenAI 高级调度器补充配置
openai_scheduler:
# 是否允许 session_hash sticky 在账号健康度恶化时临时逃逸；false 可一键回退旧行为
sticky_escape_enabled: true
# TTFT EWMA 超过该阈值（毫秒）时跳过 sticky，默认 15s，避免轻微抖动就逃逸
sticky_escape_ttft_ms: 15000
# 错误率 EWMA 超过该阈值时跳过 sticky，默认 0.5，仅在明显降级时触发
sticky_escape_error_rate: 0.5
# OpenAI HTTP upstream protocol strategy.
# OpenAI HTTP 上游协议策略（默认 HTTP/2；代理明确不兼容时可临时回退 HTTP/1.1）。
openai_http2: