wasmcloud/cmd/wasmcloud-executor/main.go

184 lines
3.9 KiB
Go
Raw Normal View History

2019-12-08 16:07:31 +00:00
package main
import (
"bytes"
"context"
"encoding/json"
"flag"
"fmt"
"io/ioutil"
"log"
"os"
"os/signal"
"sync"
"syscall"
"time"
2019-12-08 19:13:36 +00:00
"github.com/facebookgo/flagenv"
2019-12-08 16:07:31 +00:00
shell "github.com/ipfs/go-ipfs-api"
2019-12-08 19:13:36 +00:00
nats "github.com/nats-io/nats.go"
2019-12-08 16:07:31 +00:00
"github.com/perlin-network/life/exec"
"github.com/rogpeppe/go-internal/txtar"
"tulpa.dev/within/wasmcloud/cmd/internal"
2019-12-08 17:46:02 +00:00
"tulpa.dev/within/wasmcloud/executor"
2019-12-08 16:07:31 +00:00
)
var (
2019-12-08 19:13:36 +00:00
natsURL = flag.String("nats-url", nats.DefaultURL, "nats URL")
ipfsURL = flag.String("ipfs-host", "localhost:5001", "IPFS host")
2019-12-08 16:07:31 +00:00
workerCount = flag.Int("worker-count", 1, "number of wasm executor workers")
2019-12-08 17:46:02 +00:00
gasLimit = flag.Int("gas-limit", 1048576, "number of wasm instructions per execution")
ramLimit = flag.Int("ram-limit", 128, "number of wasm pages that can be used")
2019-12-08 19:13:36 +00:00
loopTimeout = flag.Duration("loop-timeout", 30*time.Second, "idle time per loop")
2019-12-08 16:07:31 +00:00
)
func main() {
2019-12-08 19:13:36 +00:00
flagenv.Parse()
flag.Parse()
2019-12-08 16:07:31 +00:00
log.SetFlags(log.LstdFlags | log.Lshortfile)
sh := shell.NewShell(*ipfsURL)
2019-12-08 19:13:36 +00:00
nc, err := nats.Connect(*natsURL)
2019-12-08 16:07:31 +00:00
if err != nil {
log.Fatal(err)
}
2019-12-08 19:13:36 +00:00
2019-12-08 16:07:31 +00:00
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
c := make(chan os.Signal, 2)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
go func() {
<-c
cancel()
2019-12-08 17:46:02 +00:00
log.Println("press ^C again to kill all of this")
<-c
os.Exit(0)
2019-12-08 16:07:31 +00:00
}()
var wg sync.WaitGroup
wg.Add(*workerCount)
for range make([]struct{}, *workerCount) {
2019-12-08 19:13:36 +00:00
go waitForNewWASM(ctx, &wg, sh, nc)
2019-12-08 16:07:31 +00:00
}
log.Printf("waiting for work on %s", internal.TopicName)
wg.Wait()
}
2019-12-08 19:13:36 +00:00
func waitForNewWASM(ctx context.Context, wg *sync.WaitGroup, sh *shell.Shell, nc *nats.Conn) {
2019-12-08 16:07:31 +00:00
defer wg.Done()
2019-12-08 19:13:36 +00:00
subsc, err := nc.QueueSubscribeSync(internal.TopicName, "workers")
if err != nil {
return
}
2019-12-08 16:07:31 +00:00
for {
select {
case <-ctx.Done():
return
default:
}
2019-12-08 19:13:36 +00:00
msg, err := subsc.NextMsg(*loopTimeout)
2019-12-08 16:07:31 +00:00
if err != nil {
2019-12-08 19:13:36 +00:00
if err == nats.ErrTimeout {
continue
}
2019-12-08 16:07:31 +00:00
log.Printf("error getting message: %v", err)
return
}
data := msg.Data
var er internal.ExecRequest
err = json.Unmarshal(data, &er)
if err != nil {
log.Printf("invalid message %s: %v", string(data), err)
continue
}
stdin := bytes.NewBuffer(er.Data)
stdout := bytes.NewBuffer(nil)
2019-12-08 17:46:02 +00:00
stderr := bytes.NewBuffer(nil)
2019-12-08 16:07:31 +00:00
logBuf := bytes.NewBuffer(nil)
bin, err := sh.Cat(er.WASMCID)
if err != nil {
log.Printf("can't get wasm: %v", err)
continue
}
wasmBin, err := ioutil.ReadAll(bin)
bin.Close()
if err != nil {
log.Printf("can't get wasm binary %s: %v", er.WASMCID, err)
continue
}
2019-12-08 17:46:02 +00:00
c := executor.Config{
VMConfig: exec.VMConfig{
GasLimit: uint64(*gasLimit),
ReturnOnGasLimitExceeded: true,
MaxMemoryPages: *ramLimit,
},
2019-12-08 16:07:31 +00:00
2019-12-08 17:46:02 +00:00
Name: er.Name,
FuncName: "cwa_main",
Env: er.Env,
Binary: wasmBin,
Stdin: stdin,
Stdout: stdout,
Stderr: stderr,
LogSink: logBuf,
2019-12-08 16:07:31 +00:00
}
2019-12-08 17:46:02 +00:00
result, err := executor.Run(c)
2019-12-08 16:07:31 +00:00
if err != nil {
2019-12-08 17:46:02 +00:00
log.Printf("can't run binary: %v", err)
2019-12-08 16:07:31 +00:00
}
arc := txtar.Archive{
2019-12-08 19:13:36 +00:00
Comment: []byte(fmt.Sprintf("%s: execution of %s (%s) at %s", er.UUID, er.Name, er.WASMCID, result.StartTime.Format(time.RFC3339))),
2019-12-08 16:07:31 +00:00
Files: []txtar.File{
{
2019-12-08 17:46:02 +00:00
Name: "logs.txt",
Data: logBuf.Bytes(),
2019-12-08 16:07:31 +00:00
},
{
Name: "stdout.txt",
Data: stdout.Bytes(),
},
{
2019-12-08 17:46:02 +00:00
Name: "stderr.txt",
Data: stderr.Bytes(),
},
result.ToFile(),
{
Name: "wasm.cid",
Data: []byte(er.WASMCID),
2019-12-08 16:07:31 +00:00
},
},
}
interm := bytes.NewBuffer(txtar.Format(&arc))
cid, err := sh.Add(interm)
if err != nil {
log.Printf("can't save execution results: %v", err)
continue
}
log.Printf("wasm module %s execution finished, logs at %s", er.WASMCID, cid)
resp := internal.ExecResponse{
WASMCID: er.WASMCID,
LogBundleCID: cid,
Logs: arc,
UUID: er.UUID,
}
repMsg, _ := json.Marshal(resp)
2019-12-08 19:13:36 +00:00
msg.Respond(repMsg)
2019-12-08 16:07:31 +00:00
}
}