This is an automated email from the ASF dual-hosted git repository.
Similarityoung pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go-pixiu.git
The following commit(s) were added to refs/heads/develop by this push:
new 51523e6a0 feat(llm): inject cooldown store via plugin and remove
shared global (#963)
51523e6a0 is described below
commit 51523e6a0e98b2979258d47ee0dbfef2e005aed9
Author: 承潜 <[email protected]>
AuthorDate: Thu Jul 23 21:02:28 2026 +0800
feat(llm): inject cooldown store via plugin and remove shared global (#963)
* feat(llm): inject cooldown store via plugin and remove shared global
The LLM proxy cooldown store was a package-level sharedCooldownStore
discovered through multi-level fallback in the filter factory and request
executor. This made ownership fuzzy and test isolation harder.
Make ownership explicit: the proxy Plugin, a process-level singleton in the
filter registry, lazily owns one cooldown store and injects it into every
FilterFactory it creates. The request executor always receives a non-nil
store from construction, so the production request path no longer needs a
fallback resolver.
ClusterManager cannot own the store without an import cycle (server does not
import the proxy package), so the Plugin is the clearly-owned process-level
runtime object the store lives on.
Closes #939
* refactor(llm): construct plugin cooldown store eagerly
The proxy Plugin is built once at init, so guarding cooldown store creation
behind sync.Once added no value over building it up front. Replace the
lazy cooldownStore() accessor with a newPlugin() constructor that creates
the store at construction time, dropping the cooldownOnce field. This also
removes the only concurrent-initialization path, so no synchronization
reasoning is needed. Tests build plugins through the constructor.
* refactor(llm): implement lazy cooldown store init with sync.Once
Address Copilot review comments:
- Add sync.Once to Plugin for lazy cooldown store initialization
- Implement Plugin.cooldownStore() accessor that safely initializes on
first call
- Update init() to use zero-value Plugin registration
- Update tests to use zero-value Plugin construction
This ensures the PR description's claim of 'lazily owns (sync.Once)' matches
the implementation, and defends against zero-value Plugin construction where
p.cooldowns would otherwise be nil.
---
pkg/filter/llm/proxy/filter.go | 57 ++++++++++++++++++-------------------
pkg/filter/llm/proxy/filter_test.go | 41 ++++++++++++++++++++++++--
2 files changed, 66 insertions(+), 32 deletions(-)
diff --git a/pkg/filter/llm/proxy/filter.go b/pkg/filter/llm/proxy/filter.go
index 394beac1c..5118a67b4 100644
--- a/pkg/filter/llm/proxy/filter.go
+++ b/pkg/filter/llm/proxy/filter.go
@@ -77,8 +77,13 @@ func init() {
}
type (
- // Plugin is the main plugin entrypoint.
- Plugin struct{}
+ // Plugin is the main plugin entrypoint. It is registered once at init
time
+ // and lives for the whole process, so it owns the cooldown store that
must
+ // outlive individual filter factory reloads.
+ Plugin struct {
+ initOnce sync.Once
+ cooldowns *cooldownStore
+ }
// FilterFactory creates filter instances.
FilterFactory struct {
@@ -133,9 +138,19 @@ type (
}
)
-// sharedCooldownStore keeps endpoint cooldowns process-wide so filter reloads
-// and multiple LLM proxy factories do not reset runtime failure state.
-var sharedCooldownStore = newCooldownStore()
+// newPlugin builds the plugin with its process-wide cooldown store, so every
+// filter factory the plugin creates shares one runtime failure state that
+// survives individual factory reloads.
+func newPlugin() *Plugin {
+ return &Plugin{cooldowns: newCooldownStore()}
+}
+
+func (p *Plugin) cooldownStore() *cooldownStore {
+ p.initOnce.Do(func() {
+ p.cooldowns = newCooldownStore()
+ })
+ return p.cooldowns
+}
func getPreferredEndpointID(hc *contexthttp.HttpContext) string {
if hc == nil || hc.Params == nil {
@@ -157,9 +172,11 @@ func (p *Plugin) Kind() string {
return Kind
}
-// CreateFilterFactory creates a new factory instance for this filter.
+// CreateFilterFactory creates a new factory instance for this filter. The
+// plugin-owned cooldown store is injected here so every factory and the
+// request executors it builds share one explicit store with no global
fallback.
func (p *Plugin) CreateFilterFactory() (filter.HttpFilterFactory, error) {
- return &FilterFactory{cfg: &Config{}}, nil
+ return &FilterFactory{cfg: &Config{}, cooldowns: p.cooldownStore()}, nil
}
// Config returns the configuration struct for the factory.
@@ -194,7 +211,7 @@ func (factory *FilterFactory) PrepareFilterChain(_
*contexthttp.HttpContext, cha
scheme: factory.cfg.Scheme,
strategy: &Strategy{},
clusterManager: server.GetClusterManager(),
- cooldowns: factory.cooldownStore(),
+ cooldowns: factory.cooldowns,
}
chain.AppendDecodeFilters(f)
return nil
@@ -417,7 +434,7 @@ func (s *Strategy) Execute(executor *RequestExecutor)
(*http.Response, error) {
}
func (executor *RequestExecutor) endpointInCooldown(endpoint *model.Endpoint)
bool {
- store := executor.cooldownStore()
+ store := executor.cooldowns
if store == nil || endpoint == nil {
return false
}
@@ -438,33 +455,13 @@ func (executor *RequestExecutor)
endpointInCooldown(endpoint *model.Endpoint) bo
}
func (executor *RequestExecutor) markEndpointCooldown(endpoint
*model.Endpoint) {
- store := executor.cooldownStore()
+ store := executor.cooldowns
if store == nil || endpoint == nil {
return
}
store.markFailure(executor.clusterName, endpoint, time.Now())
}
-func (executor *RequestExecutor) cooldownStore() *cooldownStore {
- if executor == nil {
- return nil
- }
- if executor.cooldowns != nil {
- return executor.cooldowns
- }
- if executor.filter != nil && executor.filter.cooldowns != nil {
- return executor.filter.cooldowns
- }
- return sharedCooldownStore
-}
-
-func (factory *FilterFactory) cooldownStore() *cooldownStore {
- if factory == nil || factory.cooldowns == nil {
- return sharedCooldownStore
- }
- return factory.cooldowns
-}
-
func newCooldownStore() *cooldownStore {
return &cooldownStore{
lastFailureByEndpoint: map[cooldownKey]cooldownEntry{},
diff --git a/pkg/filter/llm/proxy/filter_test.go
b/pkg/filter/llm/proxy/filter_test.go
index 05087914d..deaca8492 100644
--- a/pkg/filter/llm/proxy/filter_test.go
+++ b/pkg/filter/llm/proxy/filter_test.go
@@ -55,8 +55,8 @@ func TestFilterFactoriesShareRuntimeCooldownStore(t
*testing.T) {
return
}
- firstStore := firstFactory.(*FilterFactory).cooldownStore()
- secondStore := secondFactory.(*FilterFactory).cooldownStore()
+ firstStore := firstFactory.(*FilterFactory).cooldowns
+ secondStore := secondFactory.(*FilterFactory).cooldowns
assert.NotNil(t, firstStore)
assert.Same(t, firstStore, secondStore)
@@ -77,6 +77,43 @@ func TestFilterFactoriesShareRuntimeCooldownStore(t
*testing.T) {
assert.True(t, secondExecutor.endpointInCooldown(endpoint))
}
+func TestIndependentPluginsDoNotShareCooldownState(t *testing.T) {
+ firstPlugin := &Plugin{}
+ secondPlugin := &Plugin{}
+
+ firstFactory, err := firstPlugin.CreateFilterFactory()
+ if !assert.NoError(t, err) {
+ return
+ }
+ secondFactory, err := secondPlugin.CreateFilterFactory()
+ if !assert.NoError(t, err) {
+ return
+ }
+
+ firstStore := firstFactory.(*FilterFactory).cooldowns
+ secondStore := secondFactory.(*FilterFactory).cooldowns
+
+ assert.NotNil(t, firstStore)
+ assert.NotNil(t, secondStore)
+ assert.NotSame(t, firstStore, secondStore)
+
+ clusterName := "llm-isolated-runtime-cooldown"
+ endpoint := testLLMEndpoint("ep-1", 18189)
+ firstExecutor := &RequestExecutor{
+ clusterName: clusterName,
+ cooldowns: firstStore,
+ }
+ secondExecutor := &RequestExecutor{
+ clusterName: clusterName,
+ cooldowns: secondStore,
+ }
+
+ firstExecutor.markEndpointCooldown(endpoint)
+
+ assert.True(t, firstExecutor.endpointInCooldown(endpoint))
+ assert.False(t, secondExecutor.endpointInCooldown(endpoint))
+}
+
func
TestStrategyExecuteUsesRuntimeCooldownStateWithoutMutatingEndpointMetadata(t
*testing.T) {
clusterName := "llm-runtime-cooldown"
endpoints := []*model.Endpoint{
