mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-12 20:27:23 +00:00
fix(plugins): 分片 VM 拿不到插件配置
页面探测走的是 fanout(分片)路径,结果写回走共享 VM。分片 VM 从同一份
源码启动,但没有 state 文件、没有插件身份,所以 PLUGIN.prices(配置,不是
累计状态)在分片里是空的。
对只读累计状态的插件无所谓;对按配置分支的插件就是致命的。OpenDesign
适配器读 prices.control_key,未播种时它对每个主机都得出「未配置」并报
skip,而同一个调用走共享 VM 是好的。也就是说:生产上死、所有调共享 VM 的
测试全绿。
pool 在每次取用时推送配置(不只是首次启动时),并把配置变化也带给池中
复用的 VM;推送是拷贝,分片拿不到共享表的引用。
这条不是在写代码时想到的,是端到端验证时撞出来的:单元测试全过,隔离实例
打真上游时 vela 报「未配置」,而 control_key 明明写进去了。
(cherry picked from commit 65075b6d44)
This commit is contained in:
@ -152,6 +152,9 @@ func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
|
|||||||
w := sp.idle[n-1]
|
w := sp.idle[n-1]
|
||||||
sp.idle = sp.idle[:n-1]
|
sp.idle = sp.idle[:n-1]
|
||||||
sp.mu.Unlock()
|
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
|
return w, func() { sp.release(w) }, nil
|
||||||
}
|
}
|
||||||
if sp.limit > 0 && sp.live >= sp.limit {
|
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()
|
sp.mu.Unlock()
|
||||||
return nil, nil, err
|
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
|
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.
|
// upgrade would overwrite anyway.
|
||||||
type visibilityOverrides map[string]string
|
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
|
// SetPluginVisibility records an operator override for a plugin's audience. An
|
||||||
// empty value clears the override and restores the plugin's own default.
|
// empty value clears the override and restores the plugin's own default.
|
||||||
func (ps *Plugins) SetPluginVisibility(plugin, audience string) {
|
func (ps *Plugins) SetPluginVisibility(plugin, audience string) {
|
||||||
|
|||||||
Reference in New Issue
Block a user