From c86a9ad1da1398c16954ef40f3d02b77bb41abfa Mon Sep 17 00:00:00 2001 From: tbphp Date: Sat, 5 Sep 2026 12:58:22 +0800 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20=E6=94=AF=E6=8C=81=E5=8E=9F?= =?UTF-8?q?=E7=94=9F=E4=B8=8E=E8=BD=AC=E6=8D=A2=E8=B7=AF=E7=94=B1=E6=B7=B7?= =?UTF-8?q?=E5=90=88=E6=9D=83=E9=87=8D=E8=B0=83=E5=BA=A6=20(#570)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(scheduler): 支持原生与转换路由混合权重调度 * fix(settings): 补齐路由策略当前有效值提示 --- internal/control/group_settings_test.go | 4 + internal/control/route_inspect.go | 2 + internal/control/route_inspect_test.go | 48 ++++ internal/control/settings.go | 2 + internal/control/settings_http_test.go | 42 ++++ internal/control/settings_test.go | 52 +++++ internal/gateway/route_strategy_test.go | 82 +++++++ internal/scheduler/route_strategy_test.go | 212 ++++++++++++++++++ internal/scheduler/scheduler.go | 31 +-- internal/state/loader/loader_test.go | 4 + internal/state/runtime_settings.go | 31 +++ internal/state/runtime_settings_test.go | 28 +++ web/src/api/control/types.ts | 2 + web/src/app/resources/route-inspection.ts | 10 +- web/src/app/resources/settings.ts | 13 +- web/src/features/monitor/InspectorTab.vue | 41 ++-- .../settings/RuntimeSettingsSection.vue | 68 +++++- web/src/features/settings/settings-patch.ts | 10 +- web/src/i18n/locales/en-US/monitor.ts | 11 +- web/src/i18n/locales/en-US/settings.ts | 7 + web/src/i18n/locales/ja-JP/monitor.ts | 11 +- web/src/i18n/locales/ja-JP/settings.ts | 7 + web/src/i18n/locales/zh-CN/monitor.ts | 11 +- web/src/i18n/locales/zh-CN/settings.ts | 7 + 24 files changed, 690 insertions(+), 46 deletions(-) create mode 100644 internal/gateway/route_strategy_test.go create mode 100644 internal/scheduler/route_strategy_test.go diff --git a/internal/control/group_settings_test.go b/internal/control/group_settings_test.go index bedd1dcd..ad7a2a6d 100644 --- a/internal/control/group_settings_test.go +++ b/internal/control/group_settings_test.go @@ -460,6 +460,10 @@ func TestGroupSettingsHTTPRejectsStrictJSONAndUnauthorizedWithoutMutation(t *tes `{"unknown":true}`, `{"name":"one","name":"two"}`, `{"weight_manual":0}`, + `{"route_strategy":"weighted_mix"}`, + `{"overrides":{"route_strategy":"weighted_mix"}}`, + `{"overrides":{"route_strategy":"native_first"}}`, + `{"overrides":{"route_strategy":null}}`, } { recorder := serveGroupSettingsRequest(t, engine, http.MethodPut, path, "test-auth-key", body) if recorder.Code != http.StatusBadRequest { diff --git a/internal/control/route_inspect.go b/internal/control/route_inspect.go index 516b5328..2dc5fcc2 100644 --- a/internal/control/route_inspect.go +++ b/internal/control/route_inspect.go @@ -55,6 +55,7 @@ type routeInspectGroupResponse struct { type routeInspectResponse struct { ObservedAtMS int64 `json:"observed_at_ms"` SnapshotRevision uint64 `json:"snapshot_revision"` + RouteStrategy state.RouteStrategy `json:"route_strategy"` Protocol protocol.Protocol `json:"protocol"` Operation execution.Operation `json:"operation"` RouteRequirement execution.RouteRequirement `json:"route_requirement"` @@ -162,6 +163,7 @@ func mapRouteInspectResponse( result := routeInspectResponse{ ObservedAtMS: observedAtMS, SnapshotRevision: observation.snapshot.Revision, + RouteStrategy: observation.snapshot.Settings.RouteStrategy, Protocol: request.Protocol, Operation: explanation.Operation, RouteRequirement: explanation.RouteRequirement, diff --git a/internal/control/route_inspect_test.go b/internal/control/route_inspect_test.go index 05d13a86..cb863f8a 100644 --- a/internal/control/route_inspect_test.go +++ b/internal/control/route_inspect_test.go @@ -83,6 +83,54 @@ func routeModelValue(value *string) string { return *value } +func TestRouteInspectReportsSnapshotRouteStrategy(t *testing.T) { + t.Parallel() + initControlI18n(t) + fixture := newServiceFixture(t) + engine := gin.New() + NewServer(&config.Config{AuthKey: "test-auth-key"}, fixture.service).RegisterRoutes(engine) + + for _, strategy := range []string{"native_first", "weighted_mix"} { + t.Run(strategy, func(t *testing.T) { + settings := config.Settings{} + if strategy != "native_first" { + settings["route_strategy"] = strategy + } + snapshot, err := fixture.manager.Publish(state.CompileInput{ + ChannelRegistry: fixture.channelRegistry, + SystemSettings: settings, + Groups: []state.GroupConfig{{ + ID: 1, Name: "openai", ChannelID: channel.OpenAI, ConnectionType: "api_key", + Params: json.RawMessage(`{}`), Models: []state.ModelConfig{{ID: "model"}}, Enabled: true, + }}, + AccessKeys: []state.AccessKeyConfig{{ + ID: 10, Name: "client", KeyHash: "hash", Status: state.AccessKeyStatusActive, + }}, + }) + if err != nil { + t.Fatalf("publish route strategy: %v", err) + } + recorder := performRouteInspectRequest(engine, "test-auth-key", + `{"protocol":"openai-completions","external_model":"model","access_key_id":10}`) + got := decodeRouteInspectSuccess(t, recorder) + if got.SnapshotRevision != snapshot.Revision { + t.Fatalf("snapshot revision = %d, want %d", got.SnapshotRevision, snapshot.Revision) + } + var envelope struct { + Data struct { + RouteStrategy string `json:"route_strategy"` + } `json:"data"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil { + t.Fatal(err) + } + if envelope.Data.RouteStrategy != strategy { + t.Fatalf("route strategy = %q, want %q", envelope.Data.RouteStrategy, strategy) + } + }) + } +} + func TestRouteInspectEndpointRejectsMalformedAndInvalidRequests(t *testing.T) { t.Parallel() initControlI18n(t) diff --git a/internal/control/settings.go b/internal/control/settings.go index e30edab5..c8f59f72 100644 --- a/internal/control/settings.go +++ b/internal/control/settings.go @@ -45,6 +45,7 @@ type SettingsValuesResponse struct { ResponseHeaderRules HeaderRulesResponse `json:"response_header_rules"` InjectUsageOptions bool `json:"inject_usage_options"` RetryCount int `json:"retry_count"` + RouteStrategy state.RouteStrategy `json:"route_strategy"` BlacklistThreshold int `json:"blacklist_threshold"` AffinityEnabled bool `json:"affinity_enabled"` AffinityTTL int64 `json:"affinity_ttl"` @@ -354,6 +355,7 @@ func mapSettingsResponse( }, InjectUsageOptions: settings.InjectUsageOptions, RetryCount: settings.RetryCount, + RouteStrategy: settings.RouteStrategy, BlacklistThreshold: settings.BlacklistThreshold, AffinityEnabled: settings.AffinityEnabled, AffinityTTL: durationSeconds(settings.AffinityTTL), diff --git a/internal/control/settings_http_test.go b/internal/control/settings_http_test.go index 160d523d..1430bb4a 100644 --- a/internal/control/settings_http_test.go +++ b/internal/control/settings_http_test.go @@ -13,8 +13,50 @@ import ( "gpt-load/internal/platform/config" "gpt-load/internal/state" + "gpt-load/internal/storage/models" ) +func TestSettingsHTTPRouteStrategyRejectsInvalidValuesWithoutMutation(t *testing.T) { + t.Parallel() + initControlI18n(t) + fixture := newServiceFixture(t) + engine := gin.New() + NewServer(&config.Config{AuthKey: "test-auth-key"}, fixture.service).RegisterRoutes(engine) + + updated := serveLocalizedSettingsRequest(t, engine, http.MethodPut, "test-auth-key", "en-US", + `{"settings":{"route_strategy":"weighted_mix"}}`) + if updated.Code != http.StatusOK { + t.Fatalf("PUT weighted_mix = %d %s", updated.Code, updated.Body.String()) + } + var envelope struct { + Data SettingsResponse `json:"data"` + } + if err := json.Unmarshal(updated.Body.Bytes(), &envelope); err != nil { + t.Fatal(err) + } + if envelope.Data.Values.RouteStrategy != state.RouteStrategyWeightedMix { + t.Fatalf("response route strategy = %q", envelope.Data.Values.RouteStrategy) + } + before := fixture.manager.Current() + for _, raw := range []string{`""`, `"unknown"`, `"Native_First"`, `" weighted_mix "`, "true", "1", "[]", "{}"} { + rejected := serveLocalizedSettingsRequest(t, engine, http.MethodPut, "test-auth-key", "en-US", + `{"settings":{"route_strategy":`+raw+`,"retry_count":9}}`) + if rejected.Code != http.StatusBadRequest || !strings.Contains(rejected.Body.String(), `"code":"VALIDATION_FAILED"`) { + t.Fatalf("PUT route_strategy=%s = %d %s", raw, rejected.Code, rejected.Body.String()) + } + } + if fixture.manager.Current() != before { + t.Fatal("invalid route strategy published a new Snapshot") + } + var rows []models.SystemSetting + if err := fixture.db.Find(&rows).Error; err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Key != state.SettingRouteStrategy || rows[0].Value != `"weighted_mix"` { + t.Fatalf("invalid updates changed persisted settings: %#v", rows) + } +} + func TestSettingsHTTPLastWriteWinsWithoutPrecondition(t *testing.T) { t.Parallel() initControlI18n(t) diff --git a/internal/control/settings_test.go b/internal/control/settings_test.go index 7ac794ae..9aba5440 100644 --- a/internal/control/settings_test.go +++ b/internal/control/settings_test.go @@ -21,6 +21,7 @@ import ( "gpt-load/internal/platform/config" app_errors "gpt-load/internal/platform/errors" "gpt-load/internal/state" + stateloader "gpt-load/internal/state/loader" "gpt-load/internal/storage/models" ) @@ -77,6 +78,56 @@ func TestSettingsProxyConfigIsEncryptedMaskedAndResettable(t *testing.T) { } } +func TestUpdateSettingsRouteStrategyPersistsPublishesReloadsAndResets(t *testing.T) { + t.Parallel() + fixture := newServiceFixture(t) + for _, test := range []struct { + name string + raw json.RawMessage + strategy state.RouteStrategy + }{ + {name: "weighted mix", raw: json.RawMessage(`"weighted_mix"`), strategy: state.RouteStrategyWeightedMix}, + {name: "explicit native first", raw: json.RawMessage(`"native_first"`), strategy: state.RouteStrategyNativeFirst}, + {name: "weighted mix again", raw: json.RawMessage(`"weighted_mix"`), strategy: state.RouteStrategyWeightedMix}, + {name: "reset", raw: json.RawMessage(`null`), strategy: state.RouteStrategyNativeFirst}, + } { + t.Run(test.name, func(t *testing.T) { + before := fixture.manager.Current() + previousStrategy := before.Settings.RouteStrategy + got, err := fixture.service.UpdateSettings(t.Context(), SettingsUpdateRequest{ + Settings: map[string]json.RawMessage{state.SettingRouteStrategy: test.raw}, + }) + if err != nil { + t.Fatal(err) + } + after := fixture.manager.Current() + if got.Values.RouteStrategy != test.strategy || after.Settings.RouteStrategy != test.strategy || + after.Revision != before.Revision+1 || before.Settings.RouteStrategy != previousStrategy { + t.Fatalf("response/snapshots = %#v / %#v / %#v", got, before.Settings, after.Settings) + } + var rows []models.SystemSetting + if err := fixture.db.Where("key = ?", state.SettingRouteStrategy).Find(&rows).Error; err != nil { + t.Fatal(err) + } + if string(test.raw) == "null" { + if len(got.Overrides) != 0 || len(rows) != 0 { + t.Fatalf("reset overrides/rows = %#v / %#v", got.Overrides, rows) + } + } else if !reflect.DeepEqual(got.Overrides, []string{state.SettingRouteStrategy}) || + len(rows) != 1 || rows[0].Value != string(test.raw) { + t.Fatalf("persisted overrides/rows = %#v / %#v", got.Overrides, rows) + } + reloaded := state.NewManager() + if err := stateloader.New(fixture.db, reloaded, state.NewCredentialRegistry()).Load(t.Context()); err != nil { + t.Fatal(err) + } + if got := reloaded.Current().Settings.RouteStrategy; got != test.strategy { + t.Fatalf("reloaded route strategy = %q, want %q", got, test.strategy) + } + }) + } +} + func TestSettingsProxyRejectsInvalidConfigWithoutMutation(t *testing.T) { t.Parallel() fixture := newServiceFixture(t) @@ -110,6 +161,7 @@ func TestGetSettingsReturnsSnapshotDefaultsAndNoOverrides(t *testing.T) { if got.Values.FirstByteTimeout != 120 || got.Values.RequestTimeout != 600 || got.Values.StreamIdleTimeout != 300 || got.Values.ValidationInterval != 600 || + got.Values.RouteStrategy != state.RouteStrategyNativeFirst || got.Values.RequestLogRetentionDays != 7 || !got.Values.InjectUsageOptions { t.Fatalf("values = %#v", got.Values) } diff --git a/internal/gateway/route_strategy_test.go b/internal/gateway/route_strategy_test.go new file mode 100644 index 00000000..699538d9 --- /dev/null +++ b/internal/gateway/route_strategy_test.go @@ -0,0 +1,82 @@ +package gateway + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "gpt-load/internal/channel" + "gpt-load/internal/execution" + "gpt-load/internal/platform/config" + "gpt-load/internal/state" + "gpt-load/internal/testutil/encryptiontest" +) + +func TestHandlerUsesPublishedRouteStrategyAndPreservesSelectedRoute(t *testing.T) { + forwarder := &scriptedForwarder{results: []UpstreamResult{ + {StatusCode: http.StatusOK, Header: make(http.Header), Body: []byte(`{"ok":true}`), RequestWritten: true}, + {StatusCode: http.StatusOK, Header: make(http.Header), Body: []byte(`{"ok":true}`), RequestWritten: true}, + }} + engine, manager, registry := newHandlerTestRuntime(t, forwarder) + keyService := encryptiontest.Service(t, "handler-test-master-key") + input := state.CompileInput{ + ChannelRegistry: channel.NewRegistry(), + AccessKeys: []state.AccessKeyConfig{{ + ID: 1, Name: "client", KeyHash: keyService.Hash("gl-client"), Status: state.AccessKeyStatusActive, + }}, + } + entries := make([]state.CredentialEntry, 0, 2) + for index, target := range []struct { + channelID channel.ID + model string + weight int + }{ + {channel.OpenAI, "native-model", 1}, + {channel.Anthropic, "converted-model", 100}, + } { + id := uint(index + 1) + input.Groups = append(input.Groups, state.GroupConfig{ + ID: id, Name: string(target.channelID), ChannelID: target.channelID, ConnectionType: "api_key", + Params: json.RawMessage(`{}`), Enabled: true, WeightManual: &target.weight, + Models: []state.ModelConfig{{ID: target.model, Alias: "public"}}, + }) + credential := testCredentialConfig(id, id) + input.Credentials = append(input.Credentials, credential) + encrypted, err := keyService.Encrypt(`{"api_key":"test-upstream-key"}`) + if err != nil { + t.Fatal(err) + } + entries = append(entries, state.CredentialEntry{ + ID: id, GroupID: id, Status: credential.Status, Version: credential.Version, + IdentityGeneration: credential.IdentityGeneration, Fingerprint: credential.Fingerprint, + EncryptedValue: encrypted, + }) + } + if err := registry.ReplaceCredentials(entries); err != nil { + t.Fatal(err) + } + for index, strategy := range []state.RouteStrategy{state.RouteStrategyNativeFirst, state.RouteStrategyWeightedMix} { + input.SystemSettings = config.Settings{state.SettingRouteStrategy: string(strategy)} + if _, err := manager.Publish(input); err != nil { + t.Fatalf("publish %s: %v", strategy, err) + } + request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", + bytes.NewBufferString(`{"model":"public","messages":[{"role":"user","content":"hello"}]}`)) + request.Header.Set("Authorization", "Bearer gl-client") + recorder := httptest.NewRecorder() + engine.ServeHTTP(recorder, request) + if recorder.Code != http.StatusOK || len(forwarder.inputs) != index+1 { + t.Fatalf("%s response/attempts = %d %s / %d", strategy, recorder.Code, recorder.Body, len(forwarder.inputs)) + } + got := forwarder.inputs[index] + wantChannel, wantMode, wantModel := channel.OpenAI, execution.RouteNative, "native-model" + if strategy == state.RouteStrategyWeightedMix { + wantChannel, wantMode, wantModel = channel.Anthropic, execution.RouteConverted, "converted-model" + } + if got.ChannelID != string(wantChannel) || got.RouteMode != wantMode || got.UpstreamModelID != wantModel { + t.Fatalf("%s selected channel/mode/model = %s/%s/%s", strategy, got.ChannelID, got.RouteMode, got.UpstreamModelID) + } + } +} diff --git a/internal/scheduler/route_strategy_test.go b/internal/scheduler/route_strategy_test.go new file mode 100644 index 00000000..cfc461b3 --- /dev/null +++ b/internal/scheduler/route_strategy_test.go @@ -0,0 +1,212 @@ +package scheduler + +import ( + "errors" + "math/rand" + "slices" + "testing" + "time" + + "gpt-load/internal/channel" + "gpt-load/internal/execution" + "gpt-load/internal/platform/config" + "gpt-load/internal/protocol" + "gpt-load/internal/state" +) + +type routeStrategyRandSource int64 + +func (source routeStrategyRandSource) Int63() int64 { return int64(source) } +func (routeStrategyRandSource) Seed(int64) {} + +func setSnapshotRouteStrategy(t *testing.T, snapshot *state.ConfigSnapshot, strategy string) { + t.Helper() + settings, err := state.ResolveRuntimeSettings(config.Settings{"route_strategy": strategy}) + if err != nil { + t.Fatalf("ResolveRuntimeSettings() error = %v", err) + } + snapshot.Settings = settings +} + +func TestIteratorRouteStrategyUsesEffectiveWeightsAcrossModes(t *testing.T) { + t.Parallel() + + for _, test := range []struct { + name string + strategy string + manualWeights bool + tickets int64 + wantConverted int + }{ + {name: "native first", strategy: "native_first", manualWeights: true, tickets: 101}, + {name: "mixed manual weights", strategy: "weighted_mix", manualWeights: true, tickets: 101, wantConverted: 100}, + {name: "mixed default weights", strategy: "weighted_mix", tickets: 100, wantConverted: 50}, + } { + t.Run(test.name, func(t *testing.T) { + snapshot := channelSchedulerSnapshot(t) + setSnapshotRouteStrategy(t, snapshot, test.strategy) + if test.manualWeights { + for groupID, weight := range map[uint]int{1: 100, 2: 1} { + group := snapshot.Groups[groupID] + group.WeightManual = &weight + snapshot.Groups[groupID] = group + } + } + source := fakeCredentialSource{keys: []state.CredentialMeta{ + {ID: 11, GroupID: 1, WeightAuto: 1}, + {ID: 21, GroupID: 2, WeightAuto: 1}, + }} + converted := 0 + for ticket := range test.tickets { + selection, err := New(snapshot, source, Query{ + ClientProtocol: protocol.OpenAICompletions, + ExternalModel: modelPointer("public"), + }, rand.New(routeStrategyRandSource(ticket))).Next() + if err != nil { + t.Fatalf("ticket %d Next() error = %v", ticket, err) + } + wantMode, wantChannel, wantModel := channel.RouteNative, channel.OpenAI, "native-model" + if selection.CredentialID == 11 { + converted++ + wantMode, wantChannel, wantModel = channel.RouteConverted, channel.Anthropic, "converted-model" + } + if selection.RouteMode != wantMode || selection.ChannelID != wantChannel || + selection.ResolvedTarget.ChannelID != wantChannel || + selection.UpstreamModelID == nil || *selection.UpstreamModelID != wantModel { + t.Fatalf("ticket %d Selection lost route target: %#v", ticket, selection) + } + } + if converted != test.wantConverted { + t.Fatalf("converted selections = %d/%d, want %d", converted, test.tickets, test.wantConverted) + } + }) + } +} + +func TestIteratorWeightedMixFreezesStrategyAndPrefersEligibleConvertedCredential(t *testing.T) { + t.Parallel() + + snapshot := channelSchedulerSnapshot(t) + setSnapshotRouteStrategy(t, snapshot, "weighted_mix") + iterator := New(snapshot, fakeCredentialSource{keys: []state.CredentialMeta{ + {ID: 11, GroupID: 1}, {ID: 21, GroupID: 2}, + }}, Query{ + ClientProtocol: protocol.OpenAICompletions, + ExternalModel: modelPointer("public"), + PreferredCredentialID: 11, + }, rand.New(routeStrategyRandSource(2500))) + setSnapshotRouteStrategy(t, snapshot, "native_first") + for index, want := range []uint{11, 21} { + selection, err := iterator.Next() + if err != nil || selection.CredentialID != want { + t.Fatalf("Next() %d = (%#v, %v), want credential %d", index, selection, err, want) + } + } + if _, err := iterator.Next(); !errors.Is(err, ErrExhausted) { + t.Fatalf("Next() after exhaustion error = %v", err) + } +} + +func TestIteratorWeightedMixPreservesStoredResponsesPriority(t *testing.T) { + t.Parallel() + + for _, preferred := range []uint{0, 31} { + snapshot := responsesStoreSchedulerSnapshot(t, true) + setSnapshotRouteStrategy(t, snapshot, "weighted_mix") + weight := 100 + group := snapshot.Groups[3] + group.WeightManual = &weight + snapshot.Groups[3] = group + iterator := New(snapshot, fakeCredentialSource{keys: []state.CredentialMeta{ + {ID: 11, GroupID: 1, WeightAuto: 1}, {ID: 21, GroupID: 2, WeightAuto: 1}, + {ID: 31, GroupID: 3, WeightAuto: 1}, {ID: 41, GroupID: 4, WeightAuto: 1}, + }}, Query{ + ClientProtocol: protocol.OpenAIResponses, + Operation: execution.OperationResponsesCreate, + ResponsesStorePreference: execution.ResponsesStorePreferencePreferStored, + ExternalModel: modelPointer("gpt"), + PreferredCredentialID: preferred, + }, rand.New(routeStrategyRandSource(50))) + for index, want := range []uint{21, 31} { + selection, err := iterator.Next() + if err != nil || selection.CredentialID != want || selection.ResponsesStoreDowngraded != (index > 0) { + t.Fatalf("preferred %d Next() %d = (%#v, %v), want credential %d with preserved store semantics", preferred, index, selection, err, want) + } + } + } +} + +func TestIteratorWeightedMixKeepsNativeRequirement(t *testing.T) { + t.Parallel() + + snapshot := responsesStoreSchedulerSnapshot(t, true) + setSnapshotRouteStrategy(t, snapshot, "weighted_mix") + query := Query{ + ClientProtocol: protocol.OpenAIResponses, + Operation: execution.OperationResponsesCreate, + RouteRequirement: execution.RouteRequirementNative, + ExternalModel: modelPointer("gpt"), + PreferredCredentialID: 31, + } + if got := CandidateGroupIDsForQuery(snapshot, query); !slices.Equal(got, []uint{2}) { + t.Fatalf("CandidateGroupIDsForQuery() = %#v, want lifecycle-capable native group [2]", got) + } + iterator := New(snapshot, fakeCredentialSource{keys: []state.CredentialMeta{ + {ID: 11, GroupID: 1}, {ID: 21, GroupID: 2}, {ID: 31, GroupID: 3}, {ID: 41, GroupID: 4}, + }}, query, rand.New(zeroRandSource{})) + selection, err := iterator.Next() + if err != nil || selection.CredentialID != 21 || selection.RouteMode != channel.RouteNative { + t.Fatalf("Next() = (%#v, %v), want lifecycle-capable native credential 21", selection, err) + } + if _, err := iterator.Next(); !errors.Is(err, ErrExhausted) { + t.Fatalf("Next() after native exhaustion error = %v", err) + } +} + +func TestIteratorWeightedMixHonorsLiveHealthAndRequestExclusions(t *testing.T) { + t.Parallel() + + now := time.Unix(100, 0) + snapshot := channelSchedulerSnapshot(t) + setSnapshotRouteStrategy(t, snapshot, "weighted_mix") + registry := state.NewCredentialRegistry() + entries := make([]state.CredentialEntry, 0, 6) + zero := 0 + for _, credential := range []struct { + id, groupID uint + }{ + {11, 1}, {12, 1}, {13, 1}, {14, 1}, {21, 2}, {22, 2}, + } { + entry := state.CredentialEntry{ + ID: credential.id, GroupID: credential.groupID, Status: state.CredentialStatusActive, + Version: 1, IdentityGeneration: 1, Fingerprint: "test-fingerprint", EncryptedValue: "cipher", + } + if entry.ID == 14 { + entry.WeightManual = &zero + } + entries = append(entries, entry) + } + if err := registry.ReplaceCredentials(entries); err != nil { + t.Fatalf("ReplaceCredentials() error = %v", err) + } + allowed := map[uint]struct{}{11: {}, 12: {}, 14: {}, 21: {}, 22: {}} + iterator := newWithClock(snapshot, registry, Query{ + ClientProtocol: protocol.OpenAICompletions, + ExternalModel: modelPointer("public"), + AllowedCredentialIDs: allowed, + }, rand.New(zeroRandSource{}), func() time.Time { return now }) + allowed[13] = struct{}{} + for index, want := range []uint{11, 21} { + selection, err := iterator.Next() + if err != nil || selection.CredentialID != want { + t.Fatalf("Next() %d = (%#v, %v), want credential %d", index, selection, err, want) + } + if index == 0 && !registry.SetCooldown(12, now.Add(time.Minute)) { + t.Fatal("SetCooldown() did not find converted credential 12") + } + } + iterator.SkipGroup(2) + if _, err := iterator.Next(); !errors.Is(err, ErrExhausted) { + t.Fatalf("Next() after skip and exclusions error = %v", err) + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index b78161cf..dc4e10f6 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -52,7 +52,7 @@ type weightedCredential struct { } type candidatePool struct { - targetsByMode map[channel.RouteMode]map[uint]candidateTarget + targetsByGroup map[uint]candidateTarget groupIDsByMode map[channel.RouteMode][]uint } @@ -61,6 +61,7 @@ type Iterator struct { random *rand.Rand regular candidatePool storeDowngraded candidatePool + routeModeTiers [][]channel.RouteMode allowedCredentialIDs map[uint]struct{} preferredCredentialID uint tried map[uint]struct{} @@ -118,12 +119,16 @@ func newWithClock( random: random, regular: newCandidatePool(), storeDowngraded: newCandidatePool(), + routeModeTiers: [][]channel.RouteMode{{channel.RouteNative}, {channel.RouteConverted}}, allowedCredentialIDs: cloneAllowedCredentialIDs(query), preferredCredentialID: query.PreferredCredentialID, tried: make(map[uint]struct{}), skippedGroups: make(map[uint]struct{}), now: now, } + if snapshot != nil && snapshot.Settings.RouteStrategy == state.RouteStrategyWeightedMix { + iterator.routeModeTiers = [][]channel.RouteMode{{channel.RouteNative, channel.RouteConverted}} + } targets, staticReason := filterTargetsWithReason(snapshot, query) iterator.staticReason = staticReason for _, target := range targets { @@ -132,10 +137,7 @@ func newWithClock( pool = &iterator.storeDowngraded } mode := target.target.Mode - if pool.targetsByMode[mode] == nil { - pool.targetsByMode[mode] = make(map[uint]candidateTarget) - } - pool.targetsByMode[mode][target.target.GroupID] = target + pool.targetsByGroup[target.target.GroupID] = target pool.groupIDsByMode[mode] = append(pool.groupIDsByMode[mode], target.target.GroupID) } return iterator @@ -143,7 +145,7 @@ func newWithClock( func newCandidatePool() candidatePool { return candidatePool{ - targetsByMode: make(map[channel.RouteMode]map[uint]candidateTarget), + targetsByGroup: make(map[uint]candidateTarget), groupIDsByMode: make(map[channel.RouteMode][]uint), } } @@ -184,18 +186,21 @@ func (iterator *Iterator) weightedPoolForMode( if iterator == nil { return nil, 0 } - return iterator.weightedPoolForCandidatePool(&iterator.regular, mode, now) + return iterator.weightedPoolForCandidatePool(&iterator.regular, []channel.RouteMode{mode}, now) } func (iterator *Iterator) weightedPoolForCandidatePool( candidates *candidatePool, - mode channel.RouteMode, + modes []channel.RouteMode, now time.Time, ) ([]weightedCredential, int64) { if iterator == nil || iterator.credentials == nil { return nil, 0 } - groupIDs := candidates.groupIDsByMode[mode] + var groupIDs []uint + for _, mode := range modes { + groupIDs = append(groupIDs, candidates.groupIDsByMode[mode]...) + } if len(groupIDs) == 0 { return nil, 0 } @@ -213,7 +218,7 @@ func (iterator *Iterator) weightedPoolForCandidatePool( if _, skipped := iterator.skippedGroups[credential.GroupID]; skipped { continue } - target, ok := candidates.targetsByMode[mode][credential.GroupID] + target, ok := candidates.targetsByGroup[credential.GroupID] if !ok { continue } @@ -239,8 +244,8 @@ func (iterator *Iterator) Next() (Selection, error) { return Selection{}, ErrExhausted } for _, pool := range []*candidatePool{&iterator.regular, &iterator.storeDowngraded} { - for _, mode := range []channel.RouteMode{channel.RouteNative, channel.RouteConverted} { - weighted, total := iterator.weightedPoolForCandidatePool(pool, mode, iterator.now()) + for _, modes := range iterator.routeModeTiers { + weighted, total := iterator.weightedPoolForCandidatePool(pool, modes, iterator.now()) if total <= 0 { continue } @@ -261,7 +266,7 @@ func (iterator *Iterator) Next() (Selection, error) { } } iterator.tried[selected.ID] = struct{}{} - target := pool.targetsByMode[mode][selected.GroupID] + target := pool.targetsByGroup[selected.GroupID] return newSelection(selected, target), nil } } diff --git a/internal/state/loader/loader_test.go b/internal/state/loader/loader_test.go index 25a05e1d..9ba3e004 100644 --- a/internal/state/loader/loader_test.go +++ b/internal/state/loader/loader_test.go @@ -241,6 +241,7 @@ func TestLoaderPublishesDefaultsFromEmptyDatabase(t *testing.T) { if got.FirstByteTimeout != 120*time.Second || got.RequestTimeout != 600*time.Second || got.StreamIdleTimeout != 300*time.Second || + got.RouteStrategy != state.RouteStrategyNativeFirst || got.RequestLogRetentionDays != 7 { t.Fatalf("Settings = %#v", got) } @@ -264,6 +265,9 @@ func TestLoaderRejectsInvalidKnownPublicSystemSettingsWithoutPublishing(t *testi }{ {name: "null header rules", key: state.SettingHeaderRules, value: "null"}, {name: "fractional request timeout", key: state.SettingRequestTimeout, value: "1.5"}, + {name: "unknown route strategy", key: state.SettingRouteStrategy, value: `"unknown"`}, + {name: "null route strategy", key: state.SettingRouteStrategy, value: "null"}, + {name: "boolean route strategy", key: state.SettingRouteStrategy, value: "true"}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { diff --git a/internal/state/runtime_settings.go b/internal/state/runtime_settings.go index 2e3050f0..798dd331 100644 --- a/internal/state/runtime_settings.go +++ b/internal/state/runtime_settings.go @@ -23,6 +23,7 @@ const ( SettingResponseHeaderRules = "response_header_rules" SettingInjectUsageOptions = "inject_usage_options" SettingRetryCount = "retry_count" + SettingRouteStrategy = "route_strategy" SettingBlacklistThreshold = "blacklist_threshold" SettingAffinityEnabled = "affinity_enabled" SettingAffinityTTL = "affinity_ttl" @@ -33,6 +34,13 @@ const ( SettingParameterOverrides = "parameter_overrides" ) +type RouteStrategy string + +const ( + RouteStrategyNativeFirst RouteStrategy = "native_first" + RouteStrategyWeightedMix RouteStrategy = "weighted_mix" +) + const ( defaultRequestLogRetentionDays = 7 minRequestLogRetentionDays = 1 @@ -51,6 +59,7 @@ type RuntimeSettings struct { ResponseHeaderRules HeaderRules InjectUsageOptions bool RetryCount int + RouteStrategy RouteStrategy BlacklistThreshold int AffinityEnabled bool AffinityTTL time.Duration @@ -80,6 +89,7 @@ func DefaultRuntimeSettings() RuntimeSettings { ResponseHeaderRules: HeaderRules{Set: map[string]string{}}, InjectUsageOptions: true, RetryCount: 2, + RouteStrategy: RouteStrategyNativeFirst, BlacklistThreshold: 3, AffinityEnabled: true, AffinityTTL: time.Hour, @@ -100,6 +110,7 @@ func IsRuntimeSettingKey(key string) bool { SettingResponseHeaderRules, SettingInjectUsageOptions, SettingRetryCount, + SettingRouteStrategy, SettingBlacklistThreshold, SettingAffinityEnabled, SettingAffinityTTL, @@ -165,6 +176,12 @@ func ResolveRuntimeSettings(settings config.Settings) (RuntimeSettings, error) { return RuntimeSettings{}, err } resolved.RetryCount = count + case SettingRouteStrategy: + strategy, err := parseRouteStrategy(value) + if err != nil { + return RuntimeSettings{}, err + } + resolved.RouteStrategy = strategy case SettingBlacklistThreshold: threshold, err := nonNegativeWholeNumber(key, value) if err != nil { @@ -321,6 +338,9 @@ func ValidateRuntimeSetting(key string, value any) error { case SettingRetryCount, SettingBlacklistThreshold: _, err := nonNegativeWholeNumber(key, value) return err + case SettingRouteStrategy: + _, err := parseRouteStrategy(value) + return err case SettingAffinityEnabled: _, err := strictBoolean(key, value) return err @@ -346,6 +366,17 @@ func ValidateRuntimeSetting(key string, value any) error { } } +func parseRouteStrategy(value any) (RouteStrategy, error) { + text, ok := value.(string) + if ok { + switch strategy := RouteStrategy(text); strategy { + case RouteStrategyNativeFirst, RouteStrategyWeightedMix: + return strategy, nil + } + } + return "", fmt.Errorf("%s must be native_first or weighted_mix", SettingRouteStrategy) +} + func strictBoolean(path string, value any) (bool, error) { parsed, ok := value.(bool) if !ok { diff --git a/internal/state/runtime_settings_test.go b/internal/state/runtime_settings_test.go index 3ec545fa..fd924742 100644 --- a/internal/state/runtime_settings_test.go +++ b/internal/state/runtime_settings_test.go @@ -11,6 +11,33 @@ import ( "gpt-load/internal/platform/config" ) +func TestRouteStrategyAcceptsOnlyExplicitGlobalPolicies(t *testing.T) { + const key = "route_strategy" + if !IsRuntimeSettingKey(key) { + t.Fatal("route_strategy is not a public runtime setting") + } + for _, value := range []string{"native_first", "weighted_mix"} { + if err := ValidateRuntimeSetting(key, value); err != nil { + t.Errorf("ValidateRuntimeSetting(%q) error = %v", value, err) + } + resolved, err := ResolveRuntimeSettings(config.Settings{key: value}) + if err != nil || string(resolved.RouteStrategy) != value { + t.Errorf("ResolveRuntimeSettings(%q) = %#v, %v", value, resolved, err) + } + if _, err := ResolveGroupRuntimeSettings(DefaultRuntimeSettings(), config.Settings{key: value}); err == nil { + t.Errorf("Group accepted system-only route_strategy %q", value) + } + } + for _, value := range []any{nil, "", "unknown", "Native_First", " weighted_mix ", true, 1, []any{}, map[string]any{}} { + if err := ValidateRuntimeSetting(key, value); err == nil { + t.Errorf("ValidateRuntimeSetting(%#v) accepted invalid route strategy", value) + } + if _, err := ResolveRuntimeSettings(config.Settings{key: value}); err == nil { + t.Errorf("ResolveRuntimeSettings(%#v) accepted invalid route strategy", value) + } + } +} + func TestCompilePublishesDefaultRuntimeSettingsWithoutGroups(t *testing.T) { snapshot, err := Compile(CompileInput{}) if err != nil { @@ -29,6 +56,7 @@ func TestCompilePublishesDefaultRuntimeSettingsWithoutGroups(t *testing.T) { ResponseHeaderRules: HeaderRules{Set: map[string]string{}}, InjectUsageOptions: true, RetryCount: 2, + RouteStrategy: RouteStrategyNativeFirst, BlacklistThreshold: 3, AffinityEnabled: true, AffinityTTL: time.Hour, diff --git a/web/src/api/control/types.ts b/web/src/api/control/types.ts index 4b3c2307..6ab659ae 100644 --- a/web/src/api/control/types.ts +++ b/web/src/api/control/types.ts @@ -2,6 +2,8 @@ import type { ProtocolValue } from './protocols' export type GroupProtocol = ProtocolValue export type AccessProtocol = ProtocolValue +export const routeStrategies = ['native_first', 'weighted_mix'] as const +export type RouteStrategy = (typeof routeStrategies)[number] export type FailureCategory = | 'ok' | 'rate_limited' diff --git a/web/src/app/resources/route-inspection.ts b/web/src/app/resources/route-inspection.ts index 1392616d..dd85866f 100644 --- a/web/src/app/resources/route-inspection.ts +++ b/web/src/app/resources/route-inspection.ts @@ -1,6 +1,11 @@ import type { ApiClient } from '@/api/client' import { enabledDataProtocols } from '@/api/control/protocols' -import type { AccessKeyDto, AccessProtocol } from '@/api/control/types' +import { + routeStrategies, + type AccessKeyDto, + type AccessProtocol, + type RouteStrategy, +} from '@/api/control/types' import { InvalidResponseError } from '@/api/errors' import { @@ -87,6 +92,7 @@ export interface RouteInspectGroupDto { export interface RouteInspectResponseDto { observed_at_ms: number snapshot_revision: number + route_strategy: RouteStrategy protocol: AccessProtocol operation: RouteInspectOperation route_requirement: RouteInspectRequirement @@ -231,6 +237,7 @@ export function projectRouteInspection(value: unknown): RouteInspectResponseDto assertNoSecretLikeFields(record, [ 'observed_at_ms', 'snapshot_revision', + 'route_strategy', 'protocol', 'operation', 'route_requirement', @@ -243,6 +250,7 @@ export function projectRouteInspection(value: unknown): RouteInspectResponseDto return { observed_at_ms: projectEpochMilliseconds(record.observed_at_ms), snapshot_revision: projectSafeInteger(record.snapshot_revision, { minimum: 1 }), + route_strategy: projectEnum(record.route_strategy, routeStrategies), protocol: projectEnum(record.protocol, enabledDataProtocols), operation: projectEnum(record.operation, routeInspectOperations), route_requirement: projectEnum(record.route_requirement, routeInspectRequirements), diff --git a/web/src/app/resources/settings.ts b/web/src/app/resources/settings.ts index cf554640..1cb6166c 100644 --- a/web/src/app/resources/settings.ts +++ b/web/src/app/resources/settings.ts @@ -2,7 +2,12 @@ import { queryOptions } from '@tanstack/vue-query' import { computed, toValue, type MaybeRefOrGetter } from 'vue' import type { ApiClient } from '@/api/client' -import type { ProxyMutation, ProxyViewDto } from '@/api/control/types' +import { + routeStrategies, + type ProxyMutation, + type ProxyViewDto, + type RouteStrategy, +} from '@/api/control/types' import { InvalidResponseError } from '@/api/errors' import { controlQueryKeys } from '@/app/query-keys' @@ -11,6 +16,7 @@ import { assertNoSecretLikeFields, projectArray, projectBoolean, + projectEnum, projectRecord, projectSafeInteger, projectString, @@ -18,6 +24,7 @@ import { import { projectProxyView } from './proxy' export const runtimeSettingKeys = [ + 'route_strategy', 'first_byte_timeout', 'request_timeout', 'stream_idle_timeout', @@ -38,6 +45,7 @@ export const runtimeSettingKeys = [ export type RuntimeSettingKey = (typeof runtimeSettingKeys)[number] export type TimeoutSettingKey = Exclude< RuntimeSettingKey, + | 'route_strategy' | 'retry_count' | 'blacklist_threshold' | 'header_rules' @@ -62,6 +70,7 @@ export interface CORSConfigDto { } export interface SettingsValues { + route_strategy: RouteStrategy first_byte_timeout: number request_timeout: number stream_idle_timeout: number @@ -87,6 +96,7 @@ export interface SettingsDto { } export type SettingsPatch = Partial<{ + route_strategy: RouteStrategy | null first_byte_timeout: number | null request_timeout: number | null stream_idle_timeout: number | null @@ -186,6 +196,7 @@ export function projectSettings(value: unknown): SettingsDto { return { values: { + route_strategy: projectEnum(values.route_strategy, routeStrategies), first_byte_timeout: projectSafeInteger(values.first_byte_timeout, { minimum: 1 }), request_timeout: projectSafeInteger(values.request_timeout, { minimum: 1 }), stream_idle_timeout: projectSafeInteger(values.stream_idle_timeout, { minimum: 1 }), diff --git a/web/src/features/monitor/InspectorTab.vue b/web/src/features/monitor/InspectorTab.vue index 28d775be..f7a79e6a 100644 --- a/web/src/features/monitor/InspectorTab.vue +++ b/web/src/features/monitor/InspectorTab.vue @@ -181,6 +181,7 @@ const includedGroups = computed(() => const excludedGroups = computed(() => (observation.value?.groups ?? []).filter((group) => !group.included), ) +const weightedMix = computed(() => observation.value?.route_strategy === 'weighted_mix') const orderedIncludedGroups = computed(() => [...includedGroups.value].sort((left, right) => { const routeModeOrder = routeModePriority(left) - routeModePriority(right) @@ -199,18 +200,12 @@ const activeRouteMode = computed<'native' | 'converted' | null>(() => { } return null }) -const activeTierGroups = computed(() => - activeRouteMode.value === null - ? [] - : includedGroups.value.filter( - (group) => group.routable && group.route_mode === activeRouteMode.value, - ), -) +const activeGroups = computed(() => includedGroups.value.filter(isActiveCandidate)) const availableCredentialCount = computed(() => - activeTierGroups.value.reduce((total, group) => total + groupAvailableCredentialCount(group), 0), + activeGroups.value.reduce((total, group) => total + groupAvailableCredentialCount(group), 0), ) const totalEffectiveWeight = computed(() => - activeTierGroups.value.reduce((total, group) => total + groupEffectiveWeight(group), 0), + activeGroups.value.reduce((total, group) => total + groupEffectiveWeight(group), 0), ) function readProtocol(raw: unknown): AccessProtocol | '' { @@ -437,22 +432,27 @@ function accessKeyStatusTone(status: 'active' | 'disabled'): 'success' | 'neutra } function routeModePriority(group: RouteInspectGroupDto): number { - return group.route_mode === 'native' ? 0 : 1 + return weightedMix.value || group.route_mode === 'native' ? 0 : 1 +} + +function isActiveCandidate(group: RouteInspectGroupDto): boolean { + return group.routable && (weightedMix.value || group.route_mode === activeRouteMode.value) } function routePriorityTone(group: RouteInspectGroupDto): StatusTone { - if (!group.routable) return 'neutral' - if (group.route_mode !== activeRouteMode.value) return 'neutral' + if (!isActiveCandidate(group)) return 'neutral' return group.route_mode === 'native' ? 'success' : 'warning' } function routePriorityLabel(group: RouteInspectGroupDto): string { - return t(`monitor.inspector.groups.priority.${group.route_mode}`) + return weightedMix.value + ? t(`monitor.inspector.routeModes.${group.route_mode}`) + : t(`monitor.inspector.groups.priority.${group.route_mode}`) } function groupStatusLabel(group: RouteInspectGroupDto): string { if (!group.routable) return t('monitor.inspector.result.notRoutable') - return group.route_mode === activeRouteMode.value + return isActiveCandidate(group) ? t('monitor.inspector.groups.weightedCandidate') : t('monitor.inspector.groups.fallbackCandidate') } @@ -482,14 +482,14 @@ function groupEffectiveWeight(group: RouteInspectGroupDto): number { } function groupShare(group: RouteInspectGroupDto): number { - if (group.route_mode !== activeRouteMode.value) return 0 + if (!isActiveCandidate(group)) return 0 const total = totalEffectiveWeight.value if (total <= 0) return 0 return Math.round((groupEffectiveWeight(group) / total) * 1_000) / 10 } function groupShareLabel(group: RouteInspectGroupDto): string { - if (group.route_mode !== activeRouteMode.value) return t('monitor.inspector.groups.standbyShare') + if (!isActiveCandidate(group)) return t('monitor.inspector.groups.standbyShare') return formatPercent(groupEffectiveWeight(group), totalEffectiveWeight.value, locale.value) } @@ -637,6 +637,13 @@ onBeforeUnmount(() => {

+ + {{ + t('monitor.inspector.result.routeStrategy', { + strategy: t(`settings.runtime.routeStrategies.${observation.route_strategy}`), + }) + }} +