feat: enable external wasm runtime

Signed-off-by: Rodney Osodo <[email protected]>
absmach · Jan 10, 2025 · 562b80e · 562b80e
1 parent d2abc2a
commit 562b80e
Show file tree

Hide file tree

Showing 7 changed files with 94 additions and 42 deletions.
diff --git a/cmd/proplet/main.go b/cmd/proplet/main.go
@@ -19,15 +19,16 @@ import (
 const svcName = "proplet"
 
 type config struct {
-	LogLevel           string        `env:"PROPLET_LOG_LEVEL"           envDefault:"info"`
-	InstanceID         string        `env:"PROPLET_INSTANCE_ID"`
-	MQTTAddress        string        `env:"PROPLET_MQTT_ADDRESS"        envDefault:"tcp://localhost:1883"`
-	MQTTTimeout        time.Duration `env:"PROPLET_MQTT_TIMEOUT"        envDefault:"30s"`
-	MQTTQoS            byte          `env:"PROPLET_MQTT_QOS"            envDefault:"2"`
-	LivelinessInterval time.Duration `env:"PROPLET_LIVELINESS_INTERVAL" envDefault:"10s"`
-	ChannelID          string        `env:"PROPLET_CHANNEL_ID,notEmpty"`
-	ThingID            string        `env:"PROPLET_THING_ID,notEmpty"`
-	ThingKey           string        `env:"PROPLET_THING_KEY,notEmpty"`
+	LogLevel            string        `env:"PROPLET_LOG_LEVEL"                 envDefault:"info"`
+	InstanceID          string        `env:"PROPLET_INSTANCE_ID"`
+	MQTTAddress         string        `env:"PROPLET_MQTT_ADDRESS"              envDefault:"tcp://localhost:1883"`
+	MQTTTimeout         time.Duration `env:"PROPLET_MQTT_TIMEOUT"              envDefault:"30s"`
+	MQTTQoS             byte          `env:"PROPLET_MQTT_QOS"                  envDefault:"2"`
+	LivelinessInterval  time.Duration `env:"PROPLET_LIVELINESS_INTERVAL"       envDefault:"10s"`
+	ChannelID           string        `env:"PROPLET_CHANNEL_ID,notEmpty"`
+	ThingID             string        `env:"PROPLET_THING_ID,notEmpty"`
+	ThingKey            string        `env:"PROPLET_THING_KEY,notEmpty"`
+	ExternalWasmRuntime string        `env:"PROPLET_EXTERNAL_WASM_RUNTIME"     envDefault:""`
 }
 
 func main() {
@@ -59,7 +60,7 @@ func main() {
 
 		return
 	}
-	wazero := proplet.NewWazeroRuntime(logger, mqttPubSub, cfg.ChannelID)
+	wazero := proplet.NewWazeroRuntime(logger, mqttPubSub, cfg.ChannelID, cfg.ExternalWasmRuntime)
 
 	service, err := proplet.NewService(ctx, cfg.ChannelID, cfg.ThingID, cfg.ThingKey, cfg.LivelinessInterval, mqttPubSub, logger, wazero)
 	if err != nil {

diff --git a/manager/api/endpoint.go b/manager/api/endpoint.go
@@ -122,9 +122,6 @@ func updateTaskEndpoint(svc manager.Service) endpoint.Endpoint {
 		if !ok {
 			return taskResponse{}, errors.Join(apiutil.ErrValidation, pkgerrors.ErrInvalidData)
 		}
-		if err := req.validate(); err != nil {
-			return taskResponse{}, errors.Join(apiutil.ErrValidation, err)
-		}
 
 		task, err := svc.UpdateTask(ctx, req.Task)
 		if err != nil {

diff --git a/manager/service.go b/manager/service.go
@@ -332,24 +332,11 @@ func (svc *service) updateResultsHandler(ctx context.Context, msg map[string]int
 		return errors.New("task id is empty")
 	}
 
-	results, ok := msg["results"].([]interface{})
-	if !ok {
-		return errors.New("invalid results")
-	}
-	data := make([]uint64, len(results))
-	for i := range results {
-		r, ok := results[i].(float64)
-		if !ok {
-			return errors.New("invalid result")
-		}
-		data[i] = uint64(r)
-	}
-
 	t, err := svc.GetTask(ctx, taskID)
 	if err != nil {
 		return err
 	}
-	t.Results = data
+	t.Results = msg["results"]
 	t.State = task.Completed
 	t.UpdatedAt = time.Now()
 	t.FinishTime = time.Now()

diff --git a/proplet/requests.go b/proplet/requests.go
@@ -6,6 +6,7 @@ import (
 
 type startRequest struct {
 	ID           string
+	CLIArgs      []string
 	FunctionName string
 	WasmFile     []byte
 	imageURL     string

diff --git a/proplet/service.go b/proplet/service.go
@@ -139,6 +139,7 @@ func (p *PropletService) handleStartCommand(ctx context.Context) func(topic stri
 
 		req := startRequest{
 			ID:           payload.ID,
+			CLIArgs:      payload.CLIArgs,
 			FunctionName: payload.Name,
 			WasmFile:     payload.File,
 			imageURL:     payload.ImageURL,
@@ -151,7 +152,7 @@ func (p *PropletService) handleStartCommand(ctx context.Context) func(topic stri
 		p.logger.Info("Received start command", slog.String("app_name", req.FunctionName))
 
 		if req.WasmFile != nil {
-			if err := p.runtime.StartApp(ctx, req.WasmFile, req.ID, req.FunctionName, req.Params...); err != nil {
+			if err := p.runtime.StartApp(ctx, req.WasmFile, req.CLIArgs, req.ID, req.FunctionName, req.Params...); err != nil {
 				return err
 			}
 
@@ -178,7 +179,7 @@ func (p *PropletService) handleStartCommand(ctx context.Context) func(topic stri
 				if exists && receivedChunks == metadata.TotalChunks {
 					p.logger.Info("All chunks received, deploying app", slog.String("app_name", req.imageURL))
 					wasmBinary := assembleChunks(p.chunks[req.imageURL])
-					if err := p.runtime.StartApp(ctx, wasmBinary, req.ID, req.FunctionName, req.Params...); err != nil {
+					if err := p.runtime.StartApp(ctx, wasmBinary, req.CLIArgs, req.ID, req.FunctionName, req.Params...); err != nil {
 						p.logger.Error("Failed to start app", slog.String("app_name", req.imageURL), slog.Any("error", err))
 					}
 

diff --git a/proplet/wasm.go b/proplet/wasm.go
@@ -1,10 +1,14 @@
 package proplet
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"log/slog"
+	"os"
+	"os/exec"
+	"strconv"
 	"sync"
 
 	"github.com/absmach/propeller/pkg/mqtt"
@@ -15,28 +19,34 @@ import (
 var resultsTopic = "channels/%s/messages/control/proplet/results"
 
 type Runtime interface {
-	StartApp(ctx context.Context, wasmBinary []byte, id, functionName string, args ...uint64) error
+	StartApp(ctx context.Context, wasmBinary []byte, cliArgs []string, id, functionName string, args ...uint64) error
 	StopApp(ctx context.Context, id string) error
 }
 
 type wazeroRuntime struct {
-	mutex     sync.Mutex
-	runtimes  map[string]wazero.Runtime
-	pubsub    mqtt.PubSub
-	channelID string
-	logger    *slog.Logger
+	mutex           sync.Mutex
+	runtimes        map[string]wazero.Runtime
+	pubsub          mqtt.PubSub
+	channelID       string
+	logger          *slog.Logger
+	hostWasmRuntime string
 }
 
-func NewWazeroRuntime(logger *slog.Logger, pubsub mqtt.PubSub, channelID string) Runtime {
+func NewWazeroRuntime(logger *slog.Logger, pubsub mqtt.PubSub, channelID, hostWasmRuntime string) Runtime {
 	return &wazeroRuntime{
-		runtimes:  make(map[string]wazero.Runtime),
-		pubsub:    pubsub,
-		channelID: channelID,
-		logger:    logger,
+		runtimes:        make(map[string]wazero.Runtime),
+		pubsub:          pubsub,
+		channelID:       channelID,
+		logger:          logger,
+		hostWasmRuntime: hostWasmRuntime,
 	}
 }
 
-func (w *wazeroRuntime) StartApp(ctx context.Context, wasmBinary []byte, id, functionName string, args ...uint64) error {
+func (w *wazeroRuntime) StartApp(ctx context.Context, wasmBinary []byte, cliArgs []string, id, functionName string, args ...uint64) error {
+	if w.hostWasmRuntime != "" {
+		return w.runOnHostRuntime(ctx, wasmBinary, cliArgs, id, args...)
+	}
+
 	r := wazero.NewRuntime(ctx)
 
 	w.mutex.Lock()
@@ -104,3 +114,57 @@ func (w *wazeroRuntime) StopApp(ctx context.Context, id string) error {
 
 	return nil
 }
+
+func (w *wazeroRuntime) runOnHostRuntime(ctx context.Context, wasmBinary []byte, cliArgs []string, id string, args ...uint64) error {
+	currentDir, err := os.Getwd()
+	if err != nil {
+		return fmt.Errorf("error getting current directory: %w", err)
+	}
+	f, err := os.Create(fmt.Sprintf("%s/%s.wasm", currentDir, id))
+	if err != nil {
+		return fmt.Errorf("error creating file: %w", err)
+	}
+
+	if _, err = f.Write(wasmBinary); err != nil {
+		return fmt.Errorf("error writing to file: %w", err)
+	}
+	f.Close()
+
+	cliArgs = append(cliArgs, fmt.Sprintf("%s/%s.wasm", currentDir, id))
+	for i := range args {
+		cliArgs = append(cliArgs, strconv.FormatUint(args[i], 10))
+	}
+	cmd := exec.Command(w.hostWasmRuntime, cliArgs...)
+	results := bytes.Buffer{}
+	cmd.Stdout = &results
+
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("error starting command: %w", err)
+	}
+
+	go func(fileName string) {
+		if err := cmd.Wait(); err != nil {
+			w.logger.Error("failed to wait for command", slog.String("id", id), slog.String("error", err.Error()))
+		}
+
+		payload := map[string]interface{}{
+			"task_id": id,
+			"results": results.String(),
+		}
+
+		topic := fmt.Sprintf(resultsTopic, w.channelID)
+		if err := w.pubsub.Publish(ctx, topic, payload); err != nil {
+			w.logger.Error("failed to publish results", slog.String("id", id), slog.String("error", err.Error()))
+
+			return
+		}
+
+		if err := os.Remove(fileName); err != nil {
+			w.logger.Error("failed to remove file", slog.String("fileName", fileName), slog.String("error", err.Error()))
+		}
+
+		w.logger.Info("Finished running app", slog.String("id", id))
+	}(fmt.Sprintf("%s/%s.wasm", currentDir, id))
+
+	return nil
+}
diff --git a/task/task.go b/task/task.go
@@ -35,8 +35,9 @@ type Task struct {
 	State      State     `json:"state"`
 	ImageURL   string    `json:"image_url,omitempty"`
 	File       []byte    `json:"file,omitempty"`
+	CLIArgs    []string  `json:"cli_args"`
 	Inputs     []uint64  `json:"inputs,omitempty"`
-	Results    []uint64  `json:"results,omitempty"`
+	Results    any       `json:"results,omitempty"`
 	StartTime  time.Time `json:"start_time"`
 	FinishTime time.Time `json:"finish_time"`
 	CreatedAt  time.Time `json:"created_at"`