mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-20 17:07:59 +00:00
ConfigureConcurrency() set each adapter pool's target to the sum of
max_concurrent over its sources (108 on this deployment) and `created` only
ever went UP: once a Lua state was booted it was parked forever, so a
long-running gateway's resident state count was a high-water mark of all
traffic it had ever seen, never of what it currently needs.
Pools are now sized from three live inputs:
* the adapter's MAX CONCURRENCY (sum of max_concurrent) is a ceiling, not a
preallocation, and it sets the growth step:
growStep = clamp(ceil(maxW/8), 1, 8). A 64-wide adapter warms 8 states at
once on a spike; an 8-wide one creeps up one at a time.
* the LIVE CONNECTION COUNT (inUse, i.e. checked-out states) sets the shrink
step: shrinkStep = clamp(ceil(excess/(1+inUse)), 1, excess). With no
connections the slack collapses in a single round; a busy adapter gives up
one state per round so the hot path keeps its warm states.
* how many states already exist (created / len(idle)) decides how much room
is left to grow and how much can be reclaimed.
Batch prewarm only fires on genuine contention (a miss while every existing
state is checked out), so a single sequential caller keeps reusing one state
rather than burning a whole grow step on a cold start. A single VM-level
janitor goroutine (not one per adapter) reclaims idle states every 30s, and
shrinkGraceRounds=2 plus idleHeadroom=1 keep a gap between requests from being
mistaken for the end of a load period; any checkout resets the grace counter.
`created` now decrements on reclaim and on shutdown, and release() closes a
state outright when the ceiling was lowered underneath it, so shrinking
max_concurrent in the config gives memory back immediately instead of parking
orphans until restart.
PoolStats() exposes created/idle/in_use/waiting/max/resident/grow_step/
shrink_step/peak_in_use for the status API.
Measured on the test instance (single mock source, max_concurrent=64):
idle created=1; 50 concurrent requests -> created=10 (peak_in_use=5, ceiling
respected); after 95s of silence -> created=1. On production after deploy: 13
adapters, 1 resident Lua state total with ceilings up to 76.
1094 lines
27 KiB
Go
1094 lines
27 KiB
Go
// Package lua implements the adapter runtime: bundles/loads *.lua adapter
|
|
// scripts, exposes json/string helpers, and lets Go call the protocol
|
|
// transform functions plus a build_headers signature hook.
|
|
//
|
|
// The runtime uses LuaJIT (github.com/aarzilli/golua) so every adapter runs on
|
|
// an independent interpreter state. Because a lua.State is NOT goroutine-safe,
|
|
// each adapter owns a pool of workers (states) and a worker is checked out for
|
|
// the duration of a single hook call.
|
|
//
|
|
// Pools are ELASTIC: the sum of max_concurrent over the sources using an adapter
|
|
// is a ceiling, not a preallocation. States are booted on demand in batches
|
|
// sized by that ceiling, and reclaimed by a single background janitor at a rate
|
|
// scaled inversely to the adapter's live connection count. An idle gateway
|
|
// therefore holds close to zero Lua states regardless of how much aggregate
|
|
// concurrency is configured. See adapterPool for the sizing rules.
|
|
package lua
|
|
|
|
import (
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"embed"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
// golua exposes the LuaJIT C API. Build with -tags luajit so the cgo
|
|
// LDFLAGS resolve to -lluajit-5.1 (see golua's lua.go).
|
|
golua "github.com/aarzilli/golua/lua"
|
|
)
|
|
|
|
//go:embed adapters/*.lua
|
|
var bundledAdapters embed.FS
|
|
|
|
// adapterGlobal is the reserved global holding the adapter table after the
|
|
// worker boots the script. A reserved name avoids clashing with user code.
|
|
const adapterGlobal = "__llmsproxy_adapter"
|
|
|
|
type APIAdapter struct {
|
|
Name string `json:"name"`
|
|
Version string `json:"version"`
|
|
}
|
|
|
|
// worker wraps one LuaJIT interpreter state. Not goroutine-safe; each worker
|
|
// is used by exactly one goroutine at a time, checked out from its pool.
|
|
type worker struct {
|
|
L *golua.State
|
|
}
|
|
|
|
func (w *worker) close() {
|
|
if w.L != nil {
|
|
w.L.Close()
|
|
}
|
|
}
|
|
|
|
// staticInfo caches the immutable fields of an adapter script, extracted once
|
|
// at load time. Endpoint() and the static headers fallback read from here so
|
|
// they never churn a pooled worker.
|
|
type staticInfo struct {
|
|
name string
|
|
version string
|
|
endpoint string
|
|
headers map[string]string
|
|
}
|
|
|
|
// adapterPool owns the worker states for one adapter. Idle workers are held in
|
|
// a slice guarded by a cond; workers are created lazily on demand and reclaimed
|
|
// by the VM janitor once demand drops.
|
|
//
|
|
// Sizing is elastic rather than fixed (plan 阶段 2). Three inputs drive it:
|
|
//
|
|
// - maxW — the adapter's maximum concurrency (Σ max_concurrent of every source
|
|
// using it). It is the hard ceiling AND sets the growth step, so a hot,
|
|
// high-concurrency adapter ramps in batches while a low-concurrency one
|
|
// creeps up one state at a time.
|
|
// - created / len(idle) — how many states already exist, i.e. how much room is
|
|
// left to grow and how much is available to reclaim.
|
|
// - inUse — how many states are checked out right now, i.e. the live
|
|
// connection count for this adapter. It sets the shrink step: the fewer
|
|
// connections, the more aggressively idle states are released.
|
|
type adapterPool struct {
|
|
name string
|
|
script string
|
|
static staticInfo
|
|
|
|
mu sync.Mutex
|
|
cond *sync.Cond
|
|
idle []*worker
|
|
created int // live states: len(idle) + inUse (+ states being booted)
|
|
inUse int // states checked out right now == live connections
|
|
waiting int // goroutines blocked waiting for a free state
|
|
maxW int // hard ceiling = Σ max_concurrent of the sources using this adapter
|
|
used bool
|
|
closed bool
|
|
|
|
lastGrow time.Time
|
|
idleRounds int // consecutive janitor rounds that saw reclaimable slack
|
|
peakInUse int // high-water mark of inUse, for observability
|
|
}
|
|
|
|
const (
|
|
// growDivisor turns the adapter's max concurrency into a growth step:
|
|
// step = ceil(maxW / growDivisor), clamped to [1, growStepCap]. A source set
|
|
// at max_concurrent=8 grows one state at a time; a 64-wide adapter warms 8
|
|
// at once instead of paying 8 sequential boots on a traffic spike.
|
|
growDivisor = 8
|
|
growStepCap = 8
|
|
// growCooldown keeps a burst of misses from batching repeatedly while the
|
|
// previous batch is still booting.
|
|
growCooldown = time.Second
|
|
// residentWorkers is how many states an adapter keeps warm once it has
|
|
// served at least one request. Booting is milliseconds, but keeping one warm
|
|
// removes that from the critical path of the next request. Adapters that
|
|
// were never used keep nothing.
|
|
residentWorkers = 1
|
|
// idleHeadroom is the spare idle state kept above the live connection count,
|
|
// so the next concurrent request does not have to boot.
|
|
idleHeadroom = 1
|
|
// shrinkInterval is how often the VM janitor reclaims idle states.
|
|
shrinkInterval = 30 * time.Second
|
|
// shrinkGraceRounds is how many consecutive janitor rounds must see slack
|
|
// before anything is released, so a gap between requests is not mistaken
|
|
// for the end of a load period.
|
|
shrinkGraceRounds = 2
|
|
)
|
|
|
|
// PoolStats is a snapshot of one adapter pool's elastic sizing, surfaced by the
|
|
// status API so the algorithm is observable instead of a black box.
|
|
type PoolStats struct {
|
|
Name string `json:"name"`
|
|
Created int `json:"created"`
|
|
Idle int `json:"idle"`
|
|
InUse int `json:"in_use"`
|
|
Waiting int `json:"waiting"`
|
|
Max int `json:"max"`
|
|
Resident int `json:"resident"`
|
|
GrowStep int `json:"grow_step"`
|
|
ShrinkStep int `json:"shrink_step"`
|
|
PeakInUse int `json:"peak_in_use"`
|
|
}
|
|
|
|
func newAdapterPool(name, script string, static staticInfo) *adapterPool {
|
|
p := &adapterPool{name: name, script: script, static: static, maxW: 1}
|
|
p.cond = sync.NewCond(&p.mu)
|
|
return p
|
|
}
|
|
|
|
// setMax updates the pool's hard ceiling (Σ max_concurrent of its sources).
|
|
// Lowering it does not kill live states: the janitor reclaims the excess.
|
|
func (p *adapterPool) setMax(n int) {
|
|
if n < 1 {
|
|
n = 1
|
|
}
|
|
p.mu.Lock()
|
|
p.maxW = n
|
|
p.cond.Broadcast()
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
func ceilDiv(a, b int) int {
|
|
if b <= 0 {
|
|
return a
|
|
}
|
|
return (a + b - 1) / b
|
|
}
|
|
|
|
// growStepLocked is how many states to warm on a miss, derived from the
|
|
// adapter's maximum concurrency. Caller holds p.mu.
|
|
func (p *adapterPool) growStepLocked() int {
|
|
step := ceilDiv(p.maxW, growDivisor)
|
|
if step < 1 {
|
|
return 1
|
|
}
|
|
if step > growStepCap {
|
|
return growStepCap
|
|
}
|
|
return step
|
|
}
|
|
|
|
// residentLocked is the floor the janitor will not shrink below. Caller holds p.mu.
|
|
func (p *adapterPool) residentLocked() int {
|
|
if !p.used {
|
|
return 0
|
|
}
|
|
if residentWorkers > p.maxW {
|
|
return p.maxW
|
|
}
|
|
return residentWorkers
|
|
}
|
|
|
|
// shrinkStepLocked is how many idle states to release this round. The step is
|
|
// inversely proportional to the live connection count: with no connections the
|
|
// slack collapses in one round, while a busy adapter gives up one state at a
|
|
// time so the hot path keeps its warm states. Caller holds p.mu.
|
|
func (p *adapterPool) shrinkStepLocked() int {
|
|
excess := p.created - p.keepLocked()
|
|
if excess <= 0 {
|
|
return 0
|
|
}
|
|
step := ceilDiv(excess, 1+p.inUse)
|
|
if step > excess {
|
|
step = excess
|
|
}
|
|
if step > len(p.idle) {
|
|
step = len(p.idle)
|
|
}
|
|
if step < 0 {
|
|
return 0
|
|
}
|
|
return step
|
|
}
|
|
|
|
// keepLocked is the number of live states this pool should retain right now.
|
|
// Caller holds p.mu.
|
|
func (p *adapterPool) keepLocked() int {
|
|
keep := p.residentLocked()
|
|
if h := p.inUse + idleHeadroom; h > keep {
|
|
keep = h
|
|
}
|
|
if keep > p.maxW {
|
|
keep = p.maxW
|
|
}
|
|
return keep
|
|
}
|
|
|
|
func (p *adapterPool) stats() PoolStats {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return PoolStats{
|
|
Name: p.name,
|
|
Created: p.created,
|
|
Idle: len(p.idle),
|
|
InUse: p.inUse,
|
|
Waiting: p.waiting,
|
|
Max: p.maxW,
|
|
Resident: p.residentLocked(),
|
|
GrowStep: p.growStepLocked(),
|
|
ShrinkStep: p.shrinkStepLocked(),
|
|
PeakInUse: p.peakInUse,
|
|
}
|
|
}
|
|
|
|
// boot creates a fresh LuaJIT state, registers the shared globals and executes
|
|
// the adapter script, storing the returned adapter table in the state.
|
|
func (p *adapterPool) boot() (*worker, error) {
|
|
L := golua.NewState()
|
|
L.OpenLibs()
|
|
setupGlobals(L)
|
|
if err := L.DoString(p.script); err != nil {
|
|
L.Close()
|
|
return nil, fmt.Errorf("compile adapter %s: %w", p.name, err)
|
|
}
|
|
if L.Type(-1) != golua.LUA_TTABLE {
|
|
L.Close()
|
|
return nil, fmt.Errorf("adapter %s must return a table", p.name)
|
|
}
|
|
L.SetGlobal(adapterGlobal)
|
|
L.SetTop(0)
|
|
return &worker{L: L}, nil
|
|
}
|
|
|
|
// growPlanLocked reserves capacity for a batch prewarm and returns how many
|
|
// EXTRA states to boot in the background (the caller already reserved one for
|
|
// itself). contended says the miss happened while every existing state was
|
|
// already checked out, which is the real ramp signal: sequential traffic keeps
|
|
// reusing one warm state and must never trigger a batch, while genuinely
|
|
// concurrent traffic warms a whole step at once instead of paying one boot per
|
|
// request all the way up the ramp. Caller holds p.mu.
|
|
func (p *adapterPool) growPlanLocked(contended bool) int {
|
|
if !contended {
|
|
return 0
|
|
}
|
|
if time.Since(p.lastGrow) < growCooldown {
|
|
return 0
|
|
}
|
|
extra := p.growStepLocked() - 1
|
|
if extra <= 0 {
|
|
return 0
|
|
}
|
|
if room := p.maxW - p.created; extra > room {
|
|
extra = room
|
|
}
|
|
if extra <= 0 {
|
|
return 0
|
|
}
|
|
// Reserve immediately so concurrent acquires respect the ceiling while the
|
|
// batch is still booting.
|
|
p.created += extra
|
|
p.lastGrow = time.Now()
|
|
return extra
|
|
}
|
|
|
|
// prewarm boots n reserved states and publishes them as idle. A boot failure
|
|
// releases its reservation; it is not fatal, since the synchronous path
|
|
// surfaces adapter errors already.
|
|
func (p *adapterPool) prewarm(n int) {
|
|
for i := 0; i < n; i++ {
|
|
p.mu.Lock()
|
|
closed := p.closed
|
|
p.mu.Unlock()
|
|
if closed {
|
|
p.releaseReservation(n - i)
|
|
return
|
|
}
|
|
w, err := p.boot()
|
|
if err != nil {
|
|
p.releaseReservation(n - i)
|
|
return
|
|
}
|
|
p.mu.Lock()
|
|
if p.closed {
|
|
p.created--
|
|
p.mu.Unlock()
|
|
w.close()
|
|
p.releaseReservation(n - i - 1)
|
|
return
|
|
}
|
|
p.idle = append(p.idle, w)
|
|
p.cond.Signal()
|
|
p.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (p *adapterPool) releaseReservation(n int) {
|
|
if n <= 0 {
|
|
return
|
|
}
|
|
p.mu.Lock()
|
|
p.created -= n
|
|
if p.created < 0 {
|
|
p.created = 0
|
|
}
|
|
p.cond.Broadcast()
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
func (p *adapterPool) acquire() (*worker, error) {
|
|
p.mu.Lock()
|
|
for {
|
|
if p.closed {
|
|
p.mu.Unlock()
|
|
return nil, fmt.Errorf("adapter %s pool closed", p.name)
|
|
}
|
|
if n := len(p.idle); n > 0 {
|
|
w := p.idle[n-1]
|
|
p.idle = p.idle[:n-1]
|
|
p.checkoutLocked()
|
|
p.mu.Unlock()
|
|
return w, nil
|
|
}
|
|
if p.created < p.maxW {
|
|
// Every existing state busy on a miss = real concurrency, not a
|
|
// sequential caller reusing one warm state.
|
|
contended := p.created > 0 && p.inUse >= p.created
|
|
p.created++
|
|
extra := p.growPlanLocked(contended)
|
|
p.mu.Unlock()
|
|
if extra > 0 {
|
|
go p.prewarm(extra)
|
|
}
|
|
w, err := p.boot()
|
|
if err != nil {
|
|
p.mu.Lock()
|
|
p.created--
|
|
p.cond.Signal()
|
|
p.mu.Unlock()
|
|
return nil, err
|
|
}
|
|
p.mu.Lock()
|
|
p.checkoutLocked()
|
|
p.mu.Unlock()
|
|
return w, nil
|
|
}
|
|
p.waiting++
|
|
p.cond.Wait()
|
|
p.waiting--
|
|
}
|
|
}
|
|
|
|
// checkoutLocked accounts one state as handed out. Caller holds p.mu.
|
|
func (p *adapterPool) checkoutLocked() {
|
|
p.inUse++
|
|
p.used = true
|
|
if p.inUse > p.peakInUse {
|
|
p.peakInUse = p.inUse
|
|
}
|
|
// Live demand invalidates any pending shrink decision.
|
|
p.idleRounds = 0
|
|
}
|
|
|
|
func (p *adapterPool) release(w *worker) {
|
|
w.L.SetTop(0)
|
|
p.mu.Lock()
|
|
if p.inUse > 0 {
|
|
p.inUse--
|
|
}
|
|
if p.closed {
|
|
p.created--
|
|
p.mu.Unlock()
|
|
w.close()
|
|
return
|
|
}
|
|
// A lowered ceiling is honoured on the spot rather than waiting for the
|
|
// janitor: returning a state the pool may no longer keep closes it.
|
|
if p.created > p.maxW {
|
|
p.created--
|
|
p.cond.Signal()
|
|
p.mu.Unlock()
|
|
w.close()
|
|
return
|
|
}
|
|
p.idle = append(p.idle, w)
|
|
p.cond.Signal()
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
// reclaim releases idle states down to the current keep target, using a step
|
|
// scaled by the live connection count. It returns how many states were closed.
|
|
func (p *adapterPool) reclaim() int {
|
|
p.mu.Lock()
|
|
if p.closed {
|
|
p.mu.Unlock()
|
|
return 0
|
|
}
|
|
step := p.shrinkStepLocked()
|
|
if step <= 0 {
|
|
p.idleRounds = 0
|
|
p.mu.Unlock()
|
|
return 0
|
|
}
|
|
p.idleRounds++
|
|
if p.idleRounds < shrinkGraceRounds {
|
|
p.mu.Unlock()
|
|
return 0
|
|
}
|
|
victims := append([]*worker(nil), p.idle[len(p.idle)-step:]...)
|
|
p.idle = p.idle[:len(p.idle)-step]
|
|
p.created -= step
|
|
p.idleRounds = 0
|
|
p.mu.Unlock()
|
|
for _, w := range victims {
|
|
w.close()
|
|
}
|
|
return step
|
|
}
|
|
|
|
func (p *adapterPool) shutdown() {
|
|
p.mu.Lock()
|
|
p.closed = true
|
|
idle := p.idle
|
|
p.idle = nil
|
|
p.created -= len(idle)
|
|
p.cond.Broadcast()
|
|
p.mu.Unlock()
|
|
for _, w := range idle {
|
|
w.close()
|
|
}
|
|
}
|
|
|
|
// VM aggregates per-adapter worker pools. All exported methods are safe for
|
|
// concurrent use.
|
|
type VM struct {
|
|
mu sync.RWMutex
|
|
dir string
|
|
pools map[string]*adapterPool
|
|
|
|
// janitor drives elastic shrinking. One goroutine serves every pool (not
|
|
// one per adapter), so an idle gateway costs a single sleeping goroutine.
|
|
janitorOnce sync.Once
|
|
janitorStop chan struct{}
|
|
janitorDone chan struct{}
|
|
}
|
|
|
|
func NewVM(dir string) *VM {
|
|
return &VM{
|
|
dir: dir,
|
|
pools: map[string]*adapterPool{},
|
|
janitorStop: make(chan struct{}),
|
|
janitorDone: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// startJanitor launches the single shrink loop. Idempotent.
|
|
func (v *VM) startJanitor() {
|
|
v.janitorOnce.Do(func() {
|
|
go v.janitorLoop(shrinkInterval)
|
|
})
|
|
}
|
|
|
|
func (v *VM) janitorLoop(every time.Duration) {
|
|
defer close(v.janitorDone)
|
|
t := time.NewTicker(every)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-v.janitorStop:
|
|
return
|
|
case <-t.C:
|
|
v.reclaimIdle()
|
|
}
|
|
}
|
|
}
|
|
|
|
// reclaimIdle runs one shrink round over every pool. Exported behaviour is
|
|
// tested through ReclaimIdleNow.
|
|
func (v *VM) reclaimIdle() int {
|
|
v.mu.RLock()
|
|
pools := make([]*adapterPool, 0, len(v.pools))
|
|
for _, p := range v.pools {
|
|
pools = append(pools, p)
|
|
}
|
|
v.mu.RUnlock()
|
|
n := 0
|
|
for _, p := range pools {
|
|
n += p.reclaim()
|
|
}
|
|
return n
|
|
}
|
|
|
|
// ReclaimIdleNow forces one shrink round and reports how many worker states
|
|
// were closed. Intended for tests and for an explicit "release idle memory"
|
|
// action; normal operation relies on the janitor.
|
|
func (v *VM) ReclaimIdleNow() int { return v.reclaimIdle() }
|
|
|
|
// PoolStats returns the elastic sizing snapshot of every loaded adapter,
|
|
// sorted by name.
|
|
func (v *VM) PoolStats() []PoolStats {
|
|
v.mu.RLock()
|
|
pools := make([]*adapterPool, 0, len(v.pools))
|
|
for _, p := range v.pools {
|
|
pools = append(pools, p)
|
|
}
|
|
v.mu.RUnlock()
|
|
out := make([]PoolStats, 0, len(pools))
|
|
for _, p := range pools {
|
|
out = append(out, p.stats())
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
|
return out
|
|
}
|
|
|
|
func (v *VM) Start() error {
|
|
v.startJanitor()
|
|
if v.dir == "" {
|
|
return nil
|
|
}
|
|
// First-run seeding: a brand-new adapter dir is created and populated with
|
|
// the bundled adapters. If the dir already exists it is treated as
|
|
// authoritative and never rewritten — deleting a file there is a real delete.
|
|
firstRun := false
|
|
if _, err := os.Stat(v.dir); os.IsNotExist(err) {
|
|
firstRun = true
|
|
if err := os.MkdirAll(v.dir, 0755); err != nil {
|
|
return fmt.Errorf("mkdir adapter dir: %w", err)
|
|
}
|
|
}
|
|
if firstRun {
|
|
if err := v.writeBundledAdapters(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
entries, err := os.ReadDir(v.dir)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, e := range entries {
|
|
if filepath.Ext(e.Name()) != ".lua" {
|
|
continue
|
|
}
|
|
if err := v.LoadAdapter(filepath.Join(v.dir, e.Name())); err != nil {
|
|
fmt.Printf("[lua] preload %s: %v\n", e.Name(), err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (v *VM) Stop() {
|
|
select {
|
|
case <-v.janitorStop:
|
|
// already stopped
|
|
default:
|
|
close(v.janitorStop)
|
|
}
|
|
v.mu.Lock()
|
|
pools := v.pools
|
|
v.pools = map[string]*adapterPool{}
|
|
v.mu.Unlock()
|
|
for _, p := range pools {
|
|
p.shutdown()
|
|
}
|
|
}
|
|
|
|
// ConfigureConcurrency sets each adapter's worker CEILING to the sum of
|
|
// max_concurrent of every source using it. It is a ceiling, not a preallocation:
|
|
// states are booted on demand and reclaimed when demand drops, so a config with
|
|
// a large aggregate concurrency no longer implies a large resident pool.
|
|
// Values below one are clamped to 1.
|
|
func (v *VM) ConfigureConcurrency(adapterConcurrency map[string]int) {
|
|
v.mu.RLock()
|
|
defer v.mu.RUnlock()
|
|
for name, n := range adapterConcurrency {
|
|
if p, ok := v.pools[name]; ok {
|
|
p.setMax(n)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (v *VM) ListAdapters() []APIAdapter {
|
|
v.mu.RLock()
|
|
defer v.mu.RUnlock()
|
|
list := make([]APIAdapter, 0, len(v.pools))
|
|
for name, p := range v.pools {
|
|
list = append(list, APIAdapter{Name: name, Version: p.static.version})
|
|
}
|
|
sort.Slice(list, func(i, j int) bool { return list[i].Name < list[j].Name })
|
|
return list
|
|
}
|
|
|
|
// LoadAdapter compiles and registers an adapter script from a file.
|
|
func (v *VM) LoadAdapter(path string) error {
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read adapter: %w", err)
|
|
}
|
|
name := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
|
|
return v.LoadAdapterSource(name, string(data))
|
|
}
|
|
|
|
// LoadAdapterSource registers an adapter from source code. Replacing an
|
|
// existing adapter shuts the old pool down and re-boots lazily, carrying the
|
|
// configured ceiling over so a hot-reload does not reset elastic sizing.
|
|
func (v *VM) LoadAdapterSource(name, code string) error {
|
|
static, err := inspectScript(name, code)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
maxW := 1
|
|
v.mu.RLock()
|
|
if p, ok := v.pools[static.name]; ok {
|
|
p.mu.Lock()
|
|
maxW = p.maxW
|
|
p.mu.Unlock()
|
|
}
|
|
v.mu.RUnlock()
|
|
|
|
v.mu.Lock()
|
|
if p, ok := v.pools[static.name]; ok {
|
|
p.shutdown()
|
|
}
|
|
p := newAdapterPool(static.name, code, *static)
|
|
p.setMax(maxW)
|
|
v.pools[static.name] = p
|
|
v.mu.Unlock()
|
|
v.startJanitor()
|
|
return nil
|
|
}
|
|
|
|
// RemoveAdapter evicts an adapter and shuts its workers down.
|
|
func (v *VM) RemoveAdapter(name string) {
|
|
v.mu.Lock()
|
|
p := v.pools[name]
|
|
delete(v.pools, name)
|
|
v.mu.Unlock()
|
|
if p != nil {
|
|
p.shutdown()
|
|
}
|
|
}
|
|
|
|
func (v *VM) pool(name string) *adapterPool {
|
|
v.mu.RLock()
|
|
defer v.mu.RUnlock()
|
|
return v.pools[name]
|
|
}
|
|
|
|
func (v *VM) writeBundledAdapters() error {
|
|
// Derive the set from the embedded FS instead of a hardcoded name list so
|
|
// adding an adapter never requires maintaining a second list.
|
|
entries, err := bundledAdapters.ReadDir("adapters")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, e := range entries {
|
|
if e.IsDir() || !strings.HasSuffix(e.Name(), ".lua") {
|
|
continue
|
|
}
|
|
dst := filepath.Join(v.dir, e.Name())
|
|
if _, err := os.Stat(dst); err == nil {
|
|
continue
|
|
}
|
|
data, err := bundledAdapters.ReadFile("adapters/" + e.Name())
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if err := os.WriteFile(dst, data, 0644); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Transform runs adapter.Fn(raw) and returns the resulting string.
|
|
func (v *VM) Transform(name, fn, raw string) (string, error) {
|
|
p := v.pool(name)
|
|
if p == nil {
|
|
return "", fmt.Errorf("adapter %s not loaded", name)
|
|
}
|
|
w, err := p.acquire()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer p.release(w)
|
|
L := w.L
|
|
|
|
L.SetTop(0)
|
|
defer L.SetTop(0)
|
|
|
|
L.GetGlobal(adapterGlobal)
|
|
if L.IsNil(-1) {
|
|
return "", fmt.Errorf("adapter %s has no table", name)
|
|
}
|
|
L.GetField(-1, fn)
|
|
if L.IsNil(-1) || !L.IsFunction(-1) {
|
|
return "", fmt.Errorf("adapter %s missing %s", name, fn)
|
|
}
|
|
L.PushString(raw)
|
|
if err := L.Call(1, 1); err != nil {
|
|
return "", fmt.Errorf("%s: %w", fn, err)
|
|
}
|
|
return L.ToString(-1), nil
|
|
}
|
|
|
|
// BuildHeaders executes adapter.build_headers(meta). If the adapter defines no
|
|
// build_headers hook, it falls back to the static adapter.headers table.
|
|
func (v *VM) BuildHeaders(name string, meta map[string]interface{}) (map[string]string, error) {
|
|
p := v.pool(name)
|
|
if p == nil {
|
|
return nil, fmt.Errorf("adapter %s not loaded", name)
|
|
}
|
|
|
|
w, err := p.acquire()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer p.release(w)
|
|
L := w.L
|
|
|
|
L.SetTop(0)
|
|
defer L.SetTop(0)
|
|
|
|
L.GetGlobal(adapterGlobal)
|
|
if L.IsNil(-1) {
|
|
return nil, fmt.Errorf("adapter %s has no table", name)
|
|
}
|
|
L.GetField(-1, "build_headers")
|
|
if !L.IsFunction(-1) {
|
|
// no hook: fall back to the static headers captured at load time
|
|
p.mu.Lock()
|
|
hdr := make(map[string]string, len(p.static.headers))
|
|
for k, h := range p.static.headers {
|
|
hdr[k] = h
|
|
}
|
|
p.mu.Unlock()
|
|
return hdr, nil
|
|
}
|
|
|
|
L.SetTop(0)
|
|
L.GetGlobal(adapterGlobal)
|
|
L.GetField(-1, "build_headers")
|
|
pushGoValue(L, meta)
|
|
if err := L.Call(1, 1); err != nil {
|
|
return nil, fmt.Errorf("build_headers: %w", err)
|
|
}
|
|
if L.Type(-1) != golua.LUA_TTABLE {
|
|
return nil, fmt.Errorf("build_headers returned non-table")
|
|
}
|
|
headers := map[string]string{}
|
|
luaTableToMap(L, -1, headers)
|
|
return headers, nil
|
|
}
|
|
|
|
// TransformError executes the optional adapter.transform_error(status, body)
|
|
// hook: per-source condensing of an upstream error response into a short
|
|
// client-facing reason. ok=false means the adapter defines no hook (or it
|
|
// failed) — the caller falls back to the generic Go-side condenser.
|
|
func (v *VM) TransformError(name string, status int, body string) (reason string, ok bool, err error) {
|
|
p := v.pool(name)
|
|
if p == nil {
|
|
return "", false, fmt.Errorf("adapter %s not loaded", name)
|
|
}
|
|
w, err := p.acquire()
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
defer p.release(w)
|
|
L := w.L
|
|
|
|
L.SetTop(0)
|
|
defer L.SetTop(0)
|
|
L.GetGlobal(adapterGlobal)
|
|
if L.IsNil(-1) {
|
|
return "", false, nil
|
|
}
|
|
L.GetField(-1, "transform_error")
|
|
if !L.IsFunction(-1) {
|
|
return "", false, nil
|
|
}
|
|
L.SetTop(0)
|
|
L.GetGlobal(adapterGlobal)
|
|
L.GetField(-1, "transform_error")
|
|
L.PushInteger(int64(status))
|
|
L.PushString(body)
|
|
if callErr := L.Call(2, 1); callErr != nil {
|
|
return "", false, fmt.Errorf("transform_error: %w", callErr)
|
|
}
|
|
if L.Type(-1) != golua.LUA_TSTRING {
|
|
return "", false, nil
|
|
}
|
|
return L.ToString(-1), true, nil
|
|
}
|
|
|
|
func (v *VM) Endpoint(name string) string {
|
|
p := v.pool(name)
|
|
if p == nil {
|
|
return ""
|
|
}
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return p.static.endpoint
|
|
}
|
|
|
|
// ---- static inspection (compile-once at load time) ----
|
|
|
|
func inspectScript(name, code string) (*staticInfo, error) {
|
|
L := golua.NewState()
|
|
L.OpenLibs()
|
|
setupGlobals(L)
|
|
if err := L.DoString(code); err != nil {
|
|
L.Close()
|
|
return nil, fmt.Errorf("compile adapter: %w", err)
|
|
}
|
|
if L.Type(-1) != golua.LUA_TTABLE {
|
|
L.Close()
|
|
return nil, fmt.Errorf("adapter script must return a table")
|
|
}
|
|
info := &staticInfo{name: name, headers: map[string]string{}}
|
|
if n := readString(L, -1, "name"); n != "" {
|
|
info.name = n
|
|
}
|
|
info.version = readString(L, -1, "version")
|
|
info.endpoint = readString(L, -1, "endpoint")
|
|
if getFieldBatch(L, "headers") {
|
|
luaTableToMap(L, -1, info.headers)
|
|
L.Pop(1)
|
|
} else {
|
|
L.Pop(1)
|
|
}
|
|
L.Close()
|
|
return info, nil
|
|
}
|
|
|
|
// getFieldBatch fetches field of the table at the top of the stack and returns
|
|
// true when the fetched value (now at the top) is a table.
|
|
func getFieldBatch(L *golua.State, field string) bool {
|
|
L.GetField(-1, field)
|
|
return L.Type(-1) == golua.LUA_TTABLE
|
|
}
|
|
|
|
func readString(L *golua.State, tableIdx int, field string) string {
|
|
L.GetField(tableIdx, field)
|
|
defer L.Pop(1)
|
|
if L.Type(-1) == golua.LUA_TNIL || L.Type(-1) == golua.LUA_TNONE {
|
|
return ""
|
|
}
|
|
return normalizeString(L, -1)
|
|
}
|
|
|
|
func normalizeString(L *golua.State, idx int) string {
|
|
if L.Type(idx) == golua.LUA_TNUMBER {
|
|
n := L.ToNumber(idx)
|
|
return strconv.FormatFloat(n, 'f', -1, 64)
|
|
}
|
|
return L.ToString(idx)
|
|
}
|
|
|
|
// ---- shared globals installed into every worker state ----
|
|
|
|
// restorePcall puts the real pcall/xpcall back in place. golua's OpenLibs
|
|
// replaces them with unsafe_ variants (it captures errors via its own panic
|
|
// handler); the built-in adapters rely on pcall for soft error handling.
|
|
func restorePcall(L *golua.State) {
|
|
for _, name := range []string{"pcall", "xpcall"} {
|
|
L.GetGlobal("unsafe_" + name)
|
|
if !L.IsNil(-1) {
|
|
L.SetGlobal(name)
|
|
} else {
|
|
L.Pop(1)
|
|
}
|
|
}
|
|
}
|
|
|
|
func setupGlobals(L *golua.State) {
|
|
restorePcall(L)
|
|
buildJSONTable(L)
|
|
|
|
registerFn(L, "hmac_sha256_hex", func(L *golua.State) int {
|
|
key := L.ToString(1)
|
|
data := L.ToString(2)
|
|
m := hmac.New(sha256.New, []byte(key))
|
|
m.Write([]byte(data))
|
|
L.PushString(hex.EncodeToString(m.Sum(nil)))
|
|
return 1
|
|
})
|
|
registerFn(L, "sha256_hex", func(L *golua.State) int {
|
|
h := sha256.Sum256([]byte(L.ToString(1)))
|
|
L.PushString(hex.EncodeToString(h[:]))
|
|
return 1
|
|
})
|
|
registerFn(L, "base64_encode", func(L *golua.State) int {
|
|
L.PushString(base64.StdEncoding.EncodeToString([]byte(L.ToString(1))))
|
|
return 1
|
|
})
|
|
registerFn(L, "tohex", func(L *golua.State) int {
|
|
L.PushString(hex.EncodeToString([]byte(L.ToString(1))))
|
|
return 1
|
|
})
|
|
registerFn(L, "log", func(L *golua.State) int {
|
|
fmt.Printf("[adapter/%s] %s\n", L.ToString(1), L.ToString(2))
|
|
return 0
|
|
})
|
|
}
|
|
|
|
func buildJSONTable(L *golua.State) {
|
|
L.NewTable()
|
|
for _, name := range []string{"encode", "decode"} {
|
|
var fn golua.LuaGoFunction
|
|
switch name {
|
|
case "encode":
|
|
fn = func(L *golua.State) int {
|
|
b, err := jsonEncode(luaValueToGo(L, 1))
|
|
if err != nil {
|
|
L.PushString("null")
|
|
return 1
|
|
}
|
|
L.PushString(string(b))
|
|
return 1
|
|
}
|
|
case "decode":
|
|
fn = func(L *golua.State) int {
|
|
v, err := jsonDecode(L.ToString(1))
|
|
if err != nil {
|
|
L.PushNil()
|
|
return 1
|
|
}
|
|
pushGoValue(L, v)
|
|
return 1
|
|
}
|
|
}
|
|
L.PushGoClosure(fn)
|
|
L.SetField(-2, name)
|
|
}
|
|
L.SetGlobal("json")
|
|
}
|
|
|
|
func registerFn(L *golua.State, name string, fn golua.LuaGoFunction) {
|
|
L.PushGoClosure(fn)
|
|
L.SetGlobal(name)
|
|
}
|
|
|
|
// ---- value conversions ----
|
|
|
|
func luaValueToGo(L *golua.State, idx int) interface{} {
|
|
switch L.Type(idx) {
|
|
case golua.LUA_TSTRING:
|
|
return L.ToString(idx)
|
|
case golua.LUA_TNUMBER:
|
|
return L.ToNumber(idx)
|
|
case golua.LUA_TBOOLEAN:
|
|
return L.ToBoolean(idx)
|
|
case golua.LUA_TTABLE:
|
|
return luaTableToGoValue(L, idx)
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// luaTableToGo value converts a Lua table to either []interface{} (array 1..n)
|
|
// or map[string]interface{}.
|
|
func luaTableToGoValue(L *golua.State, idx int) interface{} {
|
|
abs := idx
|
|
if idx < 0 {
|
|
abs = L.GetTop() + idx + 1
|
|
}
|
|
m := map[string]interface{}{}
|
|
allInt, n, maxK := true, 0, 0
|
|
L.PushNil()
|
|
for L.Next(abs) != 0 {
|
|
kt := L.Type(-2)
|
|
if kt == golua.LUA_TNUMBER {
|
|
k := int(L.ToNumber(-2))
|
|
if k > maxK {
|
|
maxK = k
|
|
}
|
|
} else {
|
|
allInt = false
|
|
}
|
|
m[keyToString(L, -2)] = luaValueToGo(L, -1)
|
|
L.Pop(1)
|
|
n++
|
|
}
|
|
if allInt && maxK > 0 && n == maxK {
|
|
arr := make([]interface{}, maxK)
|
|
for i := 1; i <= maxK; i++ {
|
|
arr[i-1] = m[strconv.Itoa(i)]
|
|
}
|
|
return arr
|
|
}
|
|
return m
|
|
}
|
|
|
|
func luaTableToMap(L *golua.State, idx int, dst map[string]string) {
|
|
abs := idx
|
|
if idx < 0 {
|
|
abs = L.GetTop() + idx + 1
|
|
}
|
|
L.PushNil()
|
|
for L.Next(abs) != 0 {
|
|
dst[keyToString(L, -2)] = normalizeString(L, -1)
|
|
L.Pop(1)
|
|
}
|
|
}
|
|
|
|
// keyToString reads the stack value at idx as a string without modifying the
|
|
// original (golua's ToString mutates numeric stack values in place, which
|
|
// would break the next lua_next call), then pops the temporary copy.
|
|
func keyToString(L *golua.State, idx int) string {
|
|
L.PushValue(idx)
|
|
defer L.Pop(1)
|
|
return normalizeString(L, -1)
|
|
}
|
|
|
|
func pushGoValue(L *golua.State, v interface{}) {
|
|
switch x := v.(type) {
|
|
case nil:
|
|
L.PushNil()
|
|
case string:
|
|
L.PushString(x)
|
|
case bool:
|
|
L.PushBoolean(x)
|
|
case int:
|
|
L.PushNumber(float64(x))
|
|
case int64:
|
|
L.PushNumber(float64(x))
|
|
case float64:
|
|
L.PushNumber(x)
|
|
case []interface{}:
|
|
L.NewTable()
|
|
for i, it := range x {
|
|
pushGoValue(L, it)
|
|
L.RawSeti(-2, i+1)
|
|
}
|
|
case map[string]interface{}:
|
|
L.NewTable()
|
|
for k, it := range x {
|
|
pushGoValue(L, it)
|
|
L.SetField(-2, k)
|
|
}
|
|
default:
|
|
if b, err := json.Marshal(x); err == nil {
|
|
var iv interface{}
|
|
if json.Unmarshal(b, &iv) == nil {
|
|
pushGoValue(L, iv)
|
|
return
|
|
}
|
|
}
|
|
L.PushNil()
|
|
}
|
|
}
|
|
|
|
func jsonEncode(v interface{}) ([]byte, error) { return json.Marshal(v) }
|
|
func jsonDecode(s string) (interface{}, error) {
|
|
var v interface{}
|
|
if err := json.Unmarshal([]byte(s), &v); err != nil {
|
|
return nil, err
|
|
}
|
|
return v, nil
|
|
}
|