mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
fix(plugins): 分片 VM 拿不到插件配置
页面探测走的是 fanout(分片)路径,结果写回走共享 VM。分片 VM 从同一份 源码启动,但没有 state 文件、没有插件身份,所以 PLUGIN.prices(配置,不是 累计状态)在分片里是空的。 对只读累计状态的插件无所谓;对按配置分支的插件就是致命的。OpenDesign 适配器读 prices.control_key,未播种时它对每个主机都得出「未配置」并报 skip,而同一个调用走共享 VM 是好的。也就是说:生产上死、所有调共享 VM 的 测试全绿。 pool 在每次取用时推送配置(不只是首次启动时),并把配置变化也带给池中 复用的 VM;推送是拷贝,分片拿不到共享表的引用。 这条不是在写代码时想到的,是端到端验证时撞出来的:单元测试全过,隔离实例 打真上游时 vela 报「未配置」,而 control_key 明明写进去了。
This commit is contained in:
@ -152,6 +152,9 @@ func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
|
||||
w := sp.idle[n-1]
|
||||
sp.idle = sp.idle[:n-1]
|
||||
sp.mu.Unlock()
|
||||
// Re-seed on every checkout, not only on boot: the operator may have
|
||||
// changed the configuration since this VM was pooled.
|
||||
sp.ps.seedShardConfig(name, w)
|
||||
return w, func() { sp.release(w) }, nil
|
||||
}
|
||||
if sp.limit > 0 && sp.live >= sp.limit {
|
||||
@ -168,6 +171,9 @@ func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
|
||||
sp.mu.Unlock()
|
||||
return nil, nil, err
|
||||
}
|
||||
// A fresh shard has no state file and no plugin identity, so the shared
|
||||
// plugin's configuration has to be pushed onto it (see seedShardConfig).
|
||||
sp.ps.seedShardConfig(name, w)
|
||||
return w, func() { sp.release(w) }, nil
|
||||
}
|
||||
|
||||
|
||||
176
internal/lua/plugin_shard_config_test.go
Normal file
176
internal/lua/plugin_shard_config_test.go
Normal file
@ -0,0 +1,176 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A shard VM is booted from the same plugin SOURCE but has no state file and no
|
||||
// plugin identity, so anything a plugin reads from its `prices` table
|
||||
// (configuration, as opposed to accumulated state) is empty unless the pool puts
|
||||
// it there.
|
||||
//
|
||||
// That asymmetry is invisible until a plugin's behaviour depends on its
|
||||
// configuration. The OpenDesign adapter is the first: it reads
|
||||
// prices.control_key, and without seeding it concluded "not configured" for every
|
||||
// host and reported a skip — while the identical call through the shared VM
|
||||
// worked.
|
||||
//
|
||||
// It mattered because the two paths do different jobs: the page probes through
|
||||
// the FANOUT (shard) path and records results through the shared one. A feature
|
||||
// broken on the shard path is dead in production and green in every test that
|
||||
// calls the shared VM.
|
||||
|
||||
const shardConfigProbe = `
|
||||
local P = {}
|
||||
P.name = "cfgprobe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "read_config" }
|
||||
P.read_config = function(payload)
|
||||
local c = P.prices
|
||||
return { key = (type(c) == "table") and c.control_key or nil }, nil
|
||||
end
|
||||
return P
|
||||
`
|
||||
|
||||
func shardConfigPlugins(t *testing.T) *Plugins {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
pluginDir := filepath.Join(dir, "plugins")
|
||||
if err := os.MkdirAll(pluginDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vm := NewVM(filepath.Join(dir, "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "cfgprobe.lua"),
|
||||
[]byte(shardConfigProbe), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps := NewPlugins(vm, pluginDir)
|
||||
t.Cleanup(ps.Close)
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ps
|
||||
}
|
||||
|
||||
func configKeyOf(t *testing.T, out interface{}) string {
|
||||
t.Helper()
|
||||
m, ok := out.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("result is %T (%v), want a table", out, out)
|
||||
}
|
||||
s, _ := m["key"].(string)
|
||||
return s
|
||||
}
|
||||
|
||||
func TestShardSeesPluginConfiguration(t *testing.T) {
|
||||
ps := shardConfigPlugins(t)
|
||||
if err := ps.SetState("cfgprobe", map[string]interface{}{
|
||||
"prices": map[string]interface{}{"control_key": "secret-value"},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The shared path is the control: every pre-existing test exercises it, and
|
||||
// it passes whether or not the pool seeds anything.
|
||||
out, err := ps.Call("cfgprobe", "read_config", map[string]interface{}{})
|
||||
if err != nil {
|
||||
t.Fatalf("shared call: %v", err)
|
||||
}
|
||||
if got := configKeyOf(t, out); got != "secret-value" {
|
||||
t.Fatalf("shared VM cannot read prices.control_key (%q) — the test itself "+
|
||||
"is wrong, not the shard path", got)
|
||||
}
|
||||
|
||||
// The shard path is the one the page uses.
|
||||
outs, err := ps.ShardCall("cfgprobe", "read_config",
|
||||
[]map[string]interface{}{{}, {}, {}})
|
||||
if err != nil {
|
||||
t.Fatalf("ShardCall: %v", err)
|
||||
}
|
||||
for i, o := range outs {
|
||||
if got := configKeyOf(t, o); got != "secret-value" {
|
||||
t.Fatalf("shard %d read prices.control_key as %q, want the configured "+
|
||||
"value — a plugin that branches on its configuration behaves "+
|
||||
"differently on the path the page actually uses", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A pooled VM is reused, so the seeding has to happen on every checkout rather
|
||||
// than only on boot, or a configuration change would be invisible to the pool
|
||||
// until the process restarted.
|
||||
func TestShardConfigFollowsAConfigurationChange(t *testing.T) {
|
||||
ps := shardConfigPlugins(t)
|
||||
if err := ps.SetState("cfgprobe", map[string]interface{}{
|
||||
"prices": map[string]interface{}{"control_key": "first"},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ps.ShardCall("cfgprobe", "read_config", []map[string]interface{}{{}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ps.SetState("cfgprobe", map[string]interface{}{
|
||||
"prices": map[string]interface{}{"control_key": "second"},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
outs, err := ps.ShardCall("cfgprobe", "read_config",
|
||||
[]map[string]interface{}{{}, {}, {}, {}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i, o := range outs {
|
||||
if got := configKeyOf(t, o); got != "second" {
|
||||
t.Fatalf("shard %d still reads %q after the configuration changed to "+
|
||||
"%q — pooled VMs keep a stale configuration", i, got, "second")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Reading must not disturb the shared value. That is what makes it acceptable to
|
||||
// seed configuration into a VM whose state is otherwise throwaway, and it is why
|
||||
// the seeding is a one-way copy rather than a reference to the shared table.
|
||||
func TestShardConfigSeedingDoesNotAliasSharedState(t *testing.T) {
|
||||
ps := shardConfigPlugins(t)
|
||||
if err := ps.SetState("cfgprobe", map[string]interface{}{
|
||||
"prices": map[string]interface{}{"control_key": "keep-me"},
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
outs, err := ps.ShardCall("cfgprobe", "read_config", []map[string]interface{}{{}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := configKeyOf(t, outs[0]); got != "keep-me" {
|
||||
t.Fatalf("shard read %q", got)
|
||||
}
|
||||
out, err := ps.Call("cfgprobe", "read_config", map[string]interface{}{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := configKeyOf(t, out); got != "keep-me" {
|
||||
t.Fatalf("the shared configuration changed to %q after a shard read", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestQuotaProbesThroughTheFanoutPath is the reason the above matters. If the
|
||||
// quota page stopped probing through fanout, seeding shards would be dead code
|
||||
// and the asymmetry could return unnoticed.
|
||||
func TestQuotaProbesThroughTheFanoutPath(t *testing.T) {
|
||||
src := readShippedPlugin(t, "quota.lua")
|
||||
if !strings.Contains(src, "pluginAPI.fanout(") {
|
||||
t.Fatal("quota.lua no longer probes through the fanout (shard) path — the " +
|
||||
"page and the tests would then exercise different code, which is how " +
|
||||
"the missing shard configuration stayed hidden")
|
||||
}
|
||||
if !strings.Contains(src, "memo_record") {
|
||||
t.Fatal("quota.lua no longer records through memo_record")
|
||||
}
|
||||
}
|
||||
@ -1172,6 +1172,54 @@ func audienceAllows(want, role string) bool {
|
||||
// upgrade would overwrite anyway.
|
||||
type visibilityOverrides map[string]string
|
||||
|
||||
// seedShardConfig copies the shared plugin's `prices` table onto a shard VM.
|
||||
//
|
||||
// `prices` is CONFIGURATION, as opposed to `state` (accumulated history). A shard
|
||||
// must not write state — its copy is thrown away — but it must be able to READ
|
||||
// the operator's configuration, or a plugin that branches on its configuration
|
||||
// behaves one way through Plugins.Call and another through ShardCall.
|
||||
//
|
||||
// Found by the OpenDesign adapter, which reads prices.control_key: without this, a
|
||||
// fanout probe saw an empty table, concluded "not configured", and reported a
|
||||
// skip for every host — while the identical call through the shared VM worked.
|
||||
// The page uses the fanout path, so the feature would have been dead in
|
||||
// production and alive in every test that called the shared VM.
|
||||
//
|
||||
// Called on every checkout (boot and reuse), so a configuration change reaches
|
||||
// pooled VMs too. The push is a copy: shards get the same values, never a live
|
||||
// reference to the shared table.
|
||||
func (ps *Plugins) seedShardConfig(name string, w *worker) {
|
||||
if ps == nil || w == nil || w.L == nil {
|
||||
return
|
||||
}
|
||||
ps.mu.RLock()
|
||||
p := ps.find(name)
|
||||
ps.mu.RUnlock()
|
||||
if p == nil {
|
||||
return
|
||||
}
|
||||
p.mu.Lock()
|
||||
if p.state == nil || p.state.L == nil || p.state.L == w.L {
|
||||
p.mu.Unlock()
|
||||
return
|
||||
}
|
||||
prices := snapshotField(p.state.L, "prices")
|
||||
p.mu.Unlock()
|
||||
if prices == nil {
|
||||
return
|
||||
}
|
||||
L := w.L
|
||||
L.SetTop(0)
|
||||
defer L.SetTop(0)
|
||||
L.GetGlobal(pluginGlobal)
|
||||
if L.IsNil(-1) {
|
||||
return
|
||||
}
|
||||
plug := L.GetTop()
|
||||
pushGoValue(L, prices)
|
||||
L.SetField(plug, "prices")
|
||||
}
|
||||
|
||||
// SetPluginVisibility records an operator override for a plugin's audience. An
|
||||
// empty value clears the override and restores the plugin's own default.
|
||||
func (ps *Plugins) SetPluginVisibility(plugin, audience string) {
|
||||
|
||||
Reference in New Issue
Block a user