fix(scheduler): 修复选号耗时返回值和粘性命中率重复计数

使用命名返回值传回 defer 中记录的选号耗时;按每次选号累计任意粘性命中,保留两类独立计数。增加成功、失败及双粘性回归,不改变调度策略。
This commit is contained in:
Brisbanehuang
2026-09-12 09:03:52 -04:00
parent bdb42e22f8
commit 7e152e4279
2 changed files with 81 additions and 3 deletions
@@ -131,6 +131,7 @@ type openAIAccountSchedulerMetrics struct {
selectTotal atomic.Int64
stickyPreviousHitTotal atomic.Int64
stickySessionHitTotal atomic.Int64
stickyHitTotal atomic.Int64
loadBalanceSelectTotal atomic.Int64
accountSwitchTotal atomic.Int64
latencyMsTotal atomic.Int64
@@ -172,6 +173,9 @@ func (m *openAIAccountSchedulerMetrics) recordSelect(decision OpenAIAccountSched
if decision.StickySessionHit {
m.stickySessionHitTotal.Add(1)
}
if decision.StickyPreviousHit || decision.StickySessionHit {
m.stickyHitTotal.Add(1)
}
if decision.Layer == openAIAccountScheduleLayerLoadBalance {
m.loadBalanceSelectTotal.Add(1)
}
@@ -378,12 +382,12 @@ func newDefaultOpenAIAccountScheduler(service *OpenAIGatewayService, stats *open
func (s *defaultOpenAIAccountScheduler) Select(
ctx context.Context,
req OpenAIAccountScheduleRequest,
) (*AccountSelectionResult, OpenAIAccountScheduleDecision, error) {
) (selection *AccountSelectionResult, decision OpenAIAccountScheduleDecision, err error) {
if s != nil && s.service != nil && s.service.openAIGroupRequiresPrivacySet(ctx, req.GroupID) {
req.RequirePrivacySet = true
}
decision := OpenAIAccountScheduleDecision{}
start := time.Now()
// 命名返回值保证 defer 写入的耗时同时返回给调用方。
defer func() {
decision.LatencyMs = time.Since(start).Milliseconds()
s.metrics.recordSelect(decision)
@@ -1844,6 +1848,7 @@ func (s *defaultOpenAIAccountScheduler) SnapshotMetrics() OpenAIAccountScheduler
selectTotal := s.metrics.selectTotal.Load()
prevHit := s.metrics.stickyPreviousHitTotal.Load()
sessionHit := s.metrics.stickySessionHitTotal.Load()
stickyHit := s.metrics.stickyHitTotal.Load()
switchTotal := s.metrics.accountSwitchTotal.Load()
latencyTotal := s.metrics.latencyMsTotal.Load()
loadSkewTotal := s.metrics.loadSkewMilliTotal.Load()
@@ -1859,7 +1864,7 @@ func (s *defaultOpenAIAccountScheduler) SnapshotMetrics() OpenAIAccountScheduler
}
if selectTotal > 0 {
snapshot.SchedulerLatencyMsAvg = float64(latencyTotal) / float64(selectTotal)
snapshot.StickyHitRatio = float64(prevHit+sessionHit) / float64(selectTotal)
snapshot.StickyHitRatio = float64(stickyHit) / float64(selectTotal)
snapshot.AccountSwitchRate = float64(switchTotal) / float64(selectTotal)
snapshot.LoadSkewAvg = float64(loadSkewTotal) / 1000 / float64(selectTotal)
}
@@ -0,0 +1,73 @@
package service
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
)
type schedulerLatencyAccountRepo struct{ schedulerTestOpenAIAccountRepo }
func (r schedulerLatencyAccountRepo) ListSchedulableByPlatform(ctx context.Context, platform string) ([]Account, error) {
time.Sleep(20 * time.Millisecond)
return r.schedulerTestOpenAIAccountRepo.ListSchedulableByPlatform(ctx, platform)
}
func (r schedulerLatencyAccountRepo) ListSchedulableUngroupedByPlatform(ctx context.Context, platform string) ([]Account, error) {
return r.ListSchedulableByPlatform(ctx, platform)
}
func TestOpenAISchedulerSelectReturnsRealLatency(t *testing.T) {
resetOpenAIAdvancedSchedulerSettingCacheForTest()
t.Cleanup(resetOpenAIAdvancedSchedulerSettingCacheForTest)
for _, withAccount := range []bool{false, true} {
name := "no_available_account"
var accounts []Account
if withAccount {
name = "selected_account"
accounts = []Account{{ID: 1, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 5}}
}
t.Run(name, func(t *testing.T) {
scheduler := &defaultOpenAIAccountScheduler{
service: &OpenAIGatewayService{accountRepo: schedulerLatencyAccountRepo{schedulerTestOpenAIAccountRepo{accounts: accounts}}},
stats: newOpenAIAccountRuntimeStats(),
}
selection, decision, err := scheduler.Select(context.Background(), OpenAIAccountScheduleRequest{Platform: PlatformOpenAI})
if withAccount {
require.NoError(t, err)
require.NotNil(t, selection)
require.Equal(t, int64(1), decision.SelectedAccountID)
if selection.ReleaseFunc != nil {
t.Cleanup(selection.ReleaseFunc)
}
} else {
require.Error(t, err)
}
require.GreaterOrEqual(t, decision.LatencyMs, int64(20))
metrics := scheduler.SnapshotMetrics()
require.Equal(t, decision.LatencyMs, metrics.SchedulerLatencyMsTotal)
require.Equal(t, float64(decision.LatencyMs), metrics.SchedulerLatencyMsAvg)
})
}
}
func TestOpenAISchedulerStickyHitRatioCountsSelectionOnce(t *testing.T) {
scheduler := &defaultOpenAIAccountScheduler{stats: newOpenAIAccountRuntimeStats()}
scheduler.metrics.recordSelect(OpenAIAccountScheduleDecision{StickyPreviousHit: true, StickySessionHit: true})
snapshot := scheduler.SnapshotMetrics()
require.Equal(t, int64(1), snapshot.SelectTotal)
require.Equal(t, int64(1), snapshot.StickyPreviousHitTotal)
require.Equal(t, int64(1), snapshot.StickySessionHitTotal)
require.Equal(t, 1.0, snapshot.StickyHitRatio)
scheduler.metrics.recordSelect(OpenAIAccountScheduleDecision{StickyPreviousHit: true})
scheduler.metrics.recordSelect(OpenAIAccountScheduleDecision{StickySessionHit: true})
scheduler.metrics.recordSelect(OpenAIAccountScheduleDecision{})
snapshot = scheduler.SnapshotMetrics()
require.Equal(t, int64(4), snapshot.SelectTotal)
require.Equal(t, int64(2), snapshot.StickyPreviousHitTotal)
require.Equal(t, int64(2), snapshot.StickySessionHitTotal)
require.Equal(t, 0.75, snapshot.StickyHitRatio)
}