diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 5ad63d30f8..4a1dd2a9e8 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -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) } diff --git a/backend/internal/service/openai_account_scheduler_metrics_test.go b/backend/internal/service/openai_account_scheduler_metrics_test.go new file mode 100644 index 0000000000..c4fc7ef98d --- /dev/null +++ b/backend/internal/service/openai_account_scheduler_metrics_test.go @@ -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) +}