new plugin system
This commit is contained in:
KingKDot
@@ -2,16 +2,11 @@ package plugins
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log"
|
||||
"sync"
|
||||
|
||||
"github.com/tetratelabs/wazero"
|
||||
"github.com/tetratelabs/wazero/api"
|
||||
"github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1"
|
||||
"github.com/vmihailenco/msgpack/v5"
|
||||
|
||||
"overlord-client/cmd/agent/wire"
|
||||
)
|
||||
|
||||
@@ -26,16 +21,7 @@ type Manager struct {
|
||||
type pluginInstance struct {
|
||||
id string
|
||||
manifest PluginManifest
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
runtime wazero.Runtime
|
||||
module api.Module
|
||||
stdin io.WriteCloser
|
||||
stdout io.ReadCloser
|
||||
stderr io.ReadCloser
|
||||
enc *msgpack.Encoder
|
||||
dec *msgpack.Decoder
|
||||
encMu sync.Mutex
|
||||
native NativePlugin
|
||||
}
|
||||
|
||||
type pendingBundle struct {
|
||||
@@ -135,9 +121,9 @@ func (m *Manager) FinalizeBundle(ctx context.Context, pluginId string) error {
|
||||
return m.Load(ctx, manifest, combined)
|
||||
}
|
||||
|
||||
func (m *Manager) Load(ctx context.Context, manifest PluginManifest, wasm []byte) error {
|
||||
if len(wasm) == 0 {
|
||||
return errors.New("empty wasm payload")
|
||||
func (m *Manager) Load(ctx context.Context, manifest PluginManifest, binary []byte) error {
|
||||
if len(binary) == 0 {
|
||||
return errors.New("empty plugin binary")
|
||||
}
|
||||
pluginID := manifest.ID
|
||||
if pluginID == "" {
|
||||
@@ -146,71 +132,51 @@ func (m *Manager) Load(ctx context.Context, manifest PluginManifest, wasm []byte
|
||||
|
||||
m.mu.Lock()
|
||||
if existing, ok := m.plugins[pluginID]; ok {
|
||||
_ = existing.Close()
|
||||
existing.native.Close()
|
||||
delete(m.plugins, pluginID)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
|
||||
runtimeCtx, cancel := context.WithCancel(ctx)
|
||||
rt := wazero.NewRuntime(runtimeCtx)
|
||||
if _, err := wasi_snapshot_preview1.Instantiate(runtimeCtx, rt); err != nil {
|
||||
cancel()
|
||||
return err
|
||||
}
|
||||
|
||||
stdinR, stdinW := io.Pipe()
|
||||
stdoutR, stdoutW := io.Pipe()
|
||||
stderrR, stderrW := io.Pipe()
|
||||
|
||||
config := wazero.NewModuleConfig().
|
||||
WithStdin(stdinR).
|
||||
WithStdout(stdoutW).
|
||||
WithStderr(stderrW).
|
||||
WithStartFunctions()
|
||||
|
||||
compiled, err := rt.CompileModule(runtimeCtx, wasm)
|
||||
np, err := loadNativePlugin(binary)
|
||||
if err != nil {
|
||||
cancel()
|
||||
return err
|
||||
}
|
||||
|
||||
module, err := rt.InstantiateModule(runtimeCtx, compiled, config)
|
||||
if err != nil {
|
||||
cancel()
|
||||
return err
|
||||
}
|
||||
|
||||
pi := &pluginInstance{
|
||||
id: pluginID,
|
||||
manifest: manifest,
|
||||
ctx: runtimeCtx,
|
||||
cancel: cancel,
|
||||
runtime: rt,
|
||||
module: module,
|
||||
stdin: stdinW,
|
||||
stdout: stdoutR,
|
||||
stderr: stderrR,
|
||||
enc: msgpack.NewEncoder(stdinW),
|
||||
dec: msgpack.NewDecoder(stdoutR),
|
||||
native: np,
|
||||
}
|
||||
|
||||
goSafe("plugin readLoop", pi.cancel, func() {
|
||||
m.readLoop(pi)
|
||||
})
|
||||
goSafe("plugin readStderr", pi.cancel, func() {
|
||||
m.readStderr(pi)
|
||||
})
|
||||
|
||||
if start := module.ExportedFunction("_start"); start != nil {
|
||||
goSafe("plugin _start", pi.cancel, func() {
|
||||
if _, err := start.Call(runtimeCtx); err != nil {
|
||||
log.Printf("[plugin] %s _start error: %v", pluginID, err)
|
||||
send := func(event string, payload []byte) {
|
||||
var payloadVal interface{}
|
||||
if len(payload) > 0 {
|
||||
var parsed interface{}
|
||||
if json.Unmarshal(payload, &parsed) == nil {
|
||||
payloadVal = parsed
|
||||
} else {
|
||||
payloadVal = string(payload)
|
||||
}
|
||||
}
|
||||
err := wire.WriteMsg(context.Background(), m.writer, wire.PluginEvent{
|
||||
Type: "plugin_event",
|
||||
PluginID: pluginID,
|
||||
Event: event,
|
||||
Payload: payloadVal,
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("[plugin] %s send event error: %v", pluginID, err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := m.sendInit(pi); err != nil {
|
||||
_ = pi.Close()
|
||||
hostJSON, err := json.Marshal(m.host)
|
||||
if err != nil {
|
||||
np.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
if err := np.Load(send, hostJSON); err != nil {
|
||||
np.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -218,7 +184,7 @@ func (m *Manager) Load(ctx context.Context, manifest PluginManifest, wasm []byte
|
||||
m.plugins[pluginID] = pi
|
||||
m.mu.Unlock()
|
||||
|
||||
log.Printf("[plugin] loaded %s", pluginID)
|
||||
log.Printf("[plugin] loaded %s (native)", pluginID)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -229,15 +195,19 @@ func (m *Manager) Dispatch(ctx context.Context, pluginId, event string, payload
|
||||
if pi == nil {
|
||||
return nil
|
||||
}
|
||||
msg := PluginMessage{Type: "event", Event: event, Payload: payload}
|
||||
return m.sendMessage(ctx, pi, msg)
|
||||
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return pi.native.Event(event, data)
|
||||
}
|
||||
|
||||
func (m *Manager) Close() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for id, pi := range m.plugins {
|
||||
_ = pi.Close()
|
||||
pi.native.Close()
|
||||
delete(m.plugins, id)
|
||||
}
|
||||
}
|
||||
@@ -246,103 +216,7 @@ func (m *Manager) Unload(pluginId string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if pi, ok := m.plugins[pluginId]; ok {
|
||||
_ = pi.Close()
|
||||
pi.native.Close()
|
||||
delete(m.plugins, pluginId)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) sendInit(pi *pluginInstance) error {
|
||||
payload := map[string]interface{}{
|
||||
"manifest": pi.manifest,
|
||||
"host": m.host,
|
||||
}
|
||||
msg := PluginMessage{Type: "init", Payload: payload}
|
||||
return m.sendMessage(context.Background(), pi, msg)
|
||||
}
|
||||
|
||||
func (m *Manager) sendMessage(ctx context.Context, pi *pluginInstance, msg PluginMessage) error {
|
||||
pi.encMu.Lock()
|
||||
defer pi.encMu.Unlock()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
return pi.enc.Encode(msg)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) readLoop(pi *pluginInstance) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("[plugin] %s read loop panic: %v", pi.id, r)
|
||||
}
|
||||
}()
|
||||
for {
|
||||
var msg map[string]interface{}
|
||||
if err := pi.dec.Decode(&msg); err != nil {
|
||||
if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
|
||||
return
|
||||
}
|
||||
log.Printf("[plugin] %s read error: %v", pi.id, err)
|
||||
return
|
||||
}
|
||||
m.handlePluginMessage(pi.id, msg)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) readStderr(pi *pluginInstance) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("[plugin] %s stderr loop panic: %v", pi.id, r)
|
||||
}
|
||||
}()
|
||||
buf := make([]byte, 4096)
|
||||
for {
|
||||
n, err := pi.stderr.Read(buf)
|
||||
if n > 0 {
|
||||
log.Printf("[plugin] %s stderr: %s", pi.id, string(buf[:n]))
|
||||
}
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) handlePluginMessage(pluginId string, msg map[string]interface{}) {
|
||||
msgType, _ := msg["type"].(string)
|
||||
if msgType == "log" {
|
||||
log.Printf("[plugin] %s log: %v", pluginId, msg["payload"])
|
||||
return
|
||||
}
|
||||
if msgType == "event" {
|
||||
event, _ := msg["event"].(string)
|
||||
payload := msg["payload"]
|
||||
if event == "" {
|
||||
return
|
||||
}
|
||||
err := wire.WriteMsg(context.Background(), m.writer, wire.PluginEvent{
|
||||
Type: "plugin_event",
|
||||
PluginID: pluginId,
|
||||
Event: event,
|
||||
Payload: payload,
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("[plugin] %s send event error: %v", pluginId, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (pi *pluginInstance) Close() error {
|
||||
pi.cancel()
|
||||
_ = pi.stdin.Close()
|
||||
_ = pi.stdout.Close()
|
||||
_ = pi.stderr.Close()
|
||||
if pi.module != nil {
|
||||
_ = pi.module.Close(context.Background())
|
||||
}
|
||||
if pi.runtime != nil {
|
||||
_ = pi.runtime.Close(context.Background())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user