fix(openai): 补齐非高级调度选号决策标签

This commit is contained in:
xlplbo
2026-09-20 19:27:53 +08:00
parent d2e319b2a1
commit 8d88b66668
4 changed files with 136 additions and 12 deletions
@@ -580,6 +580,8 @@ type AccountSelectionResult struct {
Acquired bool
ReleaseFunc func()
WaitPlan *AccountWaitPlan // nil means no wait allowed
// stickySessionHit 标记账号来自会话粘性绑定命中,供非高级调度路径回填决策标签。
stickySessionHit bool
// profitGate 携带本次选号真实生效的利润门(无门为 nil)。门安装在调度栈的
// 局部 ctx 上,handler 必须经 ContextWithSelectionProfitGate 重放后才能在
// 调度栈之外做抢槽后终检与准入后粘性绑定。
@@ -2221,6 +2221,20 @@ func (s *OpenAIGatewayService) loadOpenAIGroupRequiresPrivacySet(ctx context.Con
return group != nil && group.RequirePrivacySet
}
// applyLegacySelectionDecision 把非高级调度路径的选号结果回填到决策:记录选中账号,
// 命中会话粘性时标记 session_hash 层,否则保持 load_balance。
func applyLegacySelectionDecision(decision *OpenAIAccountScheduleDecision, selection *AccountSelectionResult) {
if decision == nil || selection == nil || selection.Account == nil {
return
}
decision.SelectedAccountID = selection.Account.ID
decision.SelectedAccountType = selection.Account.Type
if selection.stickySessionHit {
decision.Layer = openAIAccountScheduleLayerSessionSticky
decision.StickySessionHit = true
}
}
func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
ctx context.Context,
groupID *int64,
@@ -2302,6 +2316,7 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
return selection, decision, nil
}
if accountSupportsOpenAICapabilities(selection.Account, requiredCapability, requiredImageCapability) {
applyLegacySelectionDecision(&decision, selection)
return selection, decision, nil
}
if selection.ReleaseFunc != nil {
@@ -2328,6 +2343,7 @@ func (s *OpenAIGatewayService) selectAccountWithSchedulerOnce(
}
if s.isOpenAIAccountTransportCompatible(selection.Account, requiredTransport) &&
accountSupportsOpenAICapabilities(selection.Account, requiredCapability, requiredImageCapability) {
applyLegacySelectionDecision(&decision, selection)
return selection, decision, nil
}
if selection.ReleaseFunc != nil {
@@ -882,25 +882,32 @@ func resolveOpenAIErrorSchedulingModel(billingModel, upstreamModel string) strin
}
func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, error) {
account, _, err := s.selectAccountForModelWithExclusionsStickyHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate)
return account, err
}
// selectAccountForModelWithExclusionsStickyHit 与 selectAccountForModelWithExclusions 相同,
// 另返回账号是否来自粘性会话命中。
func (s *OpenAIGatewayService) selectAccountForModelWithExclusionsStickyHit(ctx context.Context, groupID *int64, platform string, sessionHash string, requestedModel string, excludedIDs map[int64]struct{}, requireCompact bool, stickyAccountID int64, requiredCapability OpenAIEndpointCapability, preferLowUpstreamRate bool) (*Account, bool, error) {
platform = NormalizeOpenAICompatiblePlatform(platform)
if s.checkChannelPricingRestriction(ctx, groupID, requestedModel) {
slog.Warn("channel pricing restriction blocked request",
"group_id", derefGroupID(groupID),
"model", requestedModel)
return nil, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
return nil, false, fmt.Errorf("%w supporting model: %s (channel pricing restriction)", ErrNoAvailableAccounts, requestedModel)
}
// 1. 尝试粘性会话命中
// Try sticky session hit
if account := s.tryStickySessionHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability); account != nil {
return account, nil
return account, true, nil
}
// 2. 获取可调度的 OpenAI 账号
// Get schedulable OpenAI accounts
accounts, err := s.listSchedulableAccounts(ctx, groupID, platform)
if err != nil {
return nil, fmt.Errorf("query accounts failed: %w", err)
return nil, false, fmt.Errorf("query accounts failed: %w", err)
}
// 3. 按优先级 + LRU 选择最佳账号
@@ -908,12 +915,12 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C
selected, compactBlocked, filterStats := s.selectBestAccount(ctx, groupID, platform, accounts, requestedModel, excludedIDs, requireCompact, requiredCapability, preferLowUpstreamRate)
if selected == nil {
return nil, noAvailableOpenAISelectionError(requestedModel, compactBlocked, filterStats.summary(""))
return nil, false, noAvailableOpenAISelectionError(requestedModel, compactBlocked, filterStats.summary(""))
}
hydrated, err := s.hydrateSelectedAccount(ctx, selected)
if err != nil {
return nil, err
return nil, false, err
}
// 4. 设置粘性会话绑定(利润门下推迟到 handler 终检通过后再绑定,
@@ -923,7 +930,7 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C
_ = s.setStickySessionAccountID(ctx, groupID, sessionHash, selected.ID, openaiStickySessionTTL)
}
return hydrated, nil
return hydrated, false, nil
}
// tryStickySessionHit 尝试从粘性会话获取账号。
@@ -1133,31 +1140,34 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex
}
}
if s.concurrencyService == nil || !cfg.LoadBatchEnabled {
account, err := s.selectAccountForModelWithExclusions(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate)
account, stickyHit, err := s.selectAccountForModelWithExclusionsStickyHit(ctx, groupID, platform, sessionHash, requestedModel, excludedIDs, requireCompact, stickyAccountID, requiredCapability, preferLowUpstreamRate)
if err != nil {
return nil, err
}
result, err := s.tryAcquireAccountSlot(ctx, account.ID, account.Concurrency)
if err == nil && result != nil && result.Acquired {
return s.newAcquiredSelectionResult(ctx, account, result.ReleaseFunc)
selection, selectErr := s.newAcquiredSelectionResult(ctx, account, result.ReleaseFunc)
return markStickySessionHit(selection, stickyHit), selectErr
}
if stickyAccountID > 0 && stickyAccountID == account.ID && s.concurrencyService != nil {
waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, account.ID)
if waitingCount < cfg.StickySessionMaxWaiting {
return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
AccountID: account.ID,
MaxConcurrency: account.Concurrency,
Timeout: cfg.StickySessionWaitTimeout,
MaxWaiting: cfg.StickySessionMaxWaiting,
})
return markStickySessionHit(selection, stickyHit), selectErr
}
}
return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
AccountID: account.ID,
MaxConcurrency: account.Concurrency,
Timeout: cfg.FallbackWaitTimeout,
MaxWaiting: cfg.FallbackMaxWaiting,
})
return markStickySessionHit(selection, stickyHit), selectErr
}
accounts, err := s.listSchedulableAccounts(ctx, groupID, platform)
@@ -1211,17 +1221,18 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex
return nil, selectErr
}
_ = s.refreshStickySessionTTL(ctx, groupID, sessionHash, openaiStickySessionTTL)
return selection, nil
return markStickySessionHit(selection, true), nil
}
waitingCount, _ := s.concurrencyService.GetAccountWaitingCount(ctx, accountID)
if waitingCount < cfg.StickySessionMaxWaiting {
return s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
selection, selectErr := s.newSelectionResult(ctx, account, false, nil, &AccountWaitPlan{
AccountID: accountID,
MaxConcurrency: account.Concurrency,
Timeout: cfg.StickySessionWaitTimeout,
MaxWaiting: cfg.StickySessionMaxWaiting,
})
return markStickySessionHit(selection, true), selectErr
}
stickySpillover = true
}
@@ -1730,6 +1741,14 @@ func (s *OpenAIGatewayService) newAcquiredSelectionResult(ctx context.Context, a
return selection, err
}
// markStickySessionHit 在选号结果上记录账号是否来自会话粘性命中。
func markStickySessionHit(selection *AccountSelectionResult, hit bool) *AccountSelectionResult {
if selection != nil && hit {
selection.stickySessionHit = true
}
return selection
}
func (s *OpenAIGatewayService) schedulingConfig() config.GatewaySchedulingConfig {
if s.cfg != nil {
return s.cfg.Gateway.Scheduling
@@ -0,0 +1,87 @@
package service
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func newLegacySchedulerDecisionTestAccounts(groupID int64, secondWSEnabled bool) []Account {
return []Account{
{
ID: 38101, Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 0,
GroupIDs: []int64{groupID},
Extra: map[string]any{"openai_apikey_responses_websockets_v2_enabled": true},
},
{
ID: 38102, Platform: PlatformOpenAI, Type: AccountTypeAPIKey,
Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5,
GroupIDs: []int64{groupID},
Extra: map[string]any{"openai_apikey_responses_websockets_v2_enabled": secondWSEnabled},
},
}
}
func newLegacySchedulerDecisionTestService(accounts []Account, loadBatch bool, concurrency schedulerTestConcurrencyCache) *OpenAIGatewayService {
cfg := newSchedulerTestOpenAIWSV2Config()
cfg.Gateway.Scheduling.LoadBatchEnabled = loadBatch
return &OpenAIGatewayService{
accountRepo: schedulerTestOpenAIAccountRepo{accounts: accounts},
cache: &schedulerTestGatewayCache{},
cfg: cfg,
concurrencyService: NewConcurrencyService(concurrency),
}
}
func releaseLegacySchedulerDecisionSelection(selection *AccountSelectionResult) {
if selection != nil && selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
}
func TestLegacySchedulerDecision_StickySessionLayer(t *testing.T) {
ctx := context.Background()
groupID := int64(38100)
for _, loadBatch := range []bool{true, false} {
name := "load_batch"
if !loadBatch {
name = "priority_lru"
}
t.Run(name, func(t *testing.T) {
svc := newLegacySchedulerDecisionTestService(newLegacySchedulerDecisionTestAccounts(groupID, true), loadBatch, schedulerTestConcurrencyCache{})
sessionHash := "legacy-sticky-session"
require.NoError(t, svc.setStickySessionAccountID(ctx, &groupID, sessionHash, 38102, time.Hour))
selection, decision, err := svc.SelectAccountWithSchedulerForCapability(
ctx, &groupID, "", sessionHash, "gpt-5.1", nil,
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, false, true,
)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
releaseLegacySchedulerDecisionSelection(selection)
require.Equal(t, int64(38102), selection.Account.ID, "sticky binding must win over priority order")
require.Equal(t, openAIAccountScheduleLayerSessionSticky, decision.Layer)
require.True(t, decision.StickySessionHit)
require.False(t, decision.StickyPreviousHit)
require.Equal(t, int64(38102), decision.SelectedAccountID)
require.Equal(t, AccountTypeAPIKey, decision.SelectedAccountType)
selection, decision, err = svc.SelectAccountWithSchedulerForCapability(
ctx, &groupID, "", "", "gpt-5.1", nil,
OpenAIUpstreamTransportAny, OpenAIEndpointCapabilityChatCompletions, false, false, true,
)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotNil(t, selection.Account)
releaseLegacySchedulerDecisionSelection(selection)
require.Equal(t, int64(38101), selection.Account.ID)
require.Equal(t, openAIAccountScheduleLayerLoadBalance, decision.Layer)
require.False(t, decision.StickySessionHit)
require.Equal(t, int64(38101), decision.SelectedAccountID)
})
}
}