Merge pull request #17 from anthdm/js-resp

Js resp

cmd/wasmserver/main.go

@@ -54,6 +54,7 @@ func main() {
 	c.RegisterKind(actrs.KindRuntime, actrs.NewRuntime(store, modCache), &cluster.KindConfig{})
 	c.Engine().Spawn(actrs.NewMetric, actrs.KindMetric, actor.WithID("1"))
 	c.Engine().Spawn(actrs.NewRuntimeManager(c), actrs.KindRuntimeManager, actor.WithID("1"))
+	c.Engine().Spawn(actrs.NewRuntimeLog, actrs.KindRuntimeLog, actor.WithID("1"))
 	c.Start()
 
 	server := actrs.NewWasmServer(

examples/go/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"net/http"
 
 	raptor "github.com/anthdm/raptor/sdk"
@@ -20,5 +21,6 @@ func main() {
 	// router := chi.NewMux()
 	// router.Get("/dashboard", handleDashboard)
 	// router.Get("/login", handleLogin)
+	fmt.Println("user log")
 	raptor.Handle(http.HandlerFunc(handleLogin))
 }

examples/js/index.js

@@ -1,8 +1,32 @@
-// This should come from the official SDK. 
+// This should come from the official SDK.
 // But there is no official SDK yet, so we keep it here.
+function hexLog(s) {
+    for (let i = 0; i < s.length; i++) {
+        putstr(s.charCodeAt(i).toString(16).padStart(2, "0"))
+    }
+    putstr("0a")
+}
+
+console.log = hexLog
+
 function respond(res, status) {
-	console.log(res)
-	console.log(status)
+    var buffer = new ArrayBuffer(8);
+    var view = new DataView(buffer);
+    view.setUint32(0, status, true);
+    view.setUint32(4, res.length, true);
+
+    for (let i = 0; i < res.length; i++) {
+        putstr(res.charCodeAt(i).toString(16).padStart(2, "0"))
+    }
+    for (let i = 0; i < view.buffer.byteLength; i++) {
+        putstr(view.getUint8(i).toString(16).padStart(2, "0"));
+    }
 }
 
+console.log("user log here")
+console.log("user log here")
+console.log("user log here")
+console.log("user log here")
+console.log("user log here")
+
 respond("<h1>From my Raptor application</h1></br>some other stuff here</br>", 200)

internal/actrs/runtime.go

@@ -12,6 +12,7 @@ import (
 	"github.com/anthdm/hollywood/actor"
 	"github.com/anthdm/raptor/internal/runtime"
 	"github.com/anthdm/raptor/internal/shared"
+	"github.com/anthdm/raptor/internal/spidermonkey"
 	"github.com/anthdm/raptor/internal/storage"
 	"github.com/anthdm/raptor/internal/types"
 	"github.com/anthdm/raptor/proto"
@@ -39,6 +40,7 @@ type Runtime struct {
 	runtime      *runtime.Runtime
 	repeat       actor.SendRepeater
 	stdout       *bytes.Buffer
+	script       []byte
 }
 
 func NewRuntime(store storage.Store, cache storage.ModCacher) actor.Producer {
@@ -79,7 +81,7 @@ func (r *Runtime) Receive(c *actor.Context) {
 
 func (r *Runtime) initialize(msg *proto.HTTPRequest) error {
 	r.deploymentID = uuid.MustParse(msg.DeploymentID)
-	// TODO: this could be coming from a Redis cache instead of Postres.
+	// TODO: this could be coming from a Redis cache instead of Postgres.
 	// Maybe only the blob. Not sure...
 	deploy, err := r.store.GetDeployment(r.deploymentID)
 	if err != nil {
@@ -95,10 +97,18 @@ func (r *Runtime) initialize(msg *proto.HTTPRequest) error {
 	args := runtime.Args{
 		Cache:        modCache,
 		DeploymentID: deploy.ID,
-		Blob:         deploy.Blob,
 		Engine:       msg.Runtime,
 		Stdout:       r.stdout,
 	}
+
+	switch args.Engine {
+	case "js":
+		r.script = deploy.Blob
+		args.Blob = spidermonkey.WasmBlob
+	default:
+		args.Blob = deploy.Blob
+	}
+
 	run, err := runtime.New(context.Background(), args)
 	if err != nil {
 		return err
@@ -118,9 +128,9 @@ func (r *Runtime) handleHTTPRequest(ctx *actor.Context, msg *proto.HTTPRequest)
 		return
 	}
 
-	var args []string = nil
+	args := []string{}
 	if msg.Runtime == "js" {
-		args = []string{"", "-e", string(r.runtime.Blob())}
+		args = []string{"", "-e", string(r.script)}
 	}
 
 	req := bytes.NewReader(b)
@@ -130,7 +140,7 @@ func (r *Runtime) handleHTTPRequest(ctx *actor.Context, msg *proto.HTTPRequest)
 		return
 	}
 
-	res, status, err := shared.ParseRuntimeHTTPResponse(r.stdout.String())
+	logs, res, status, err := shared.ParseStdout(msg.Runtime, r.stdout)
 	if err != nil {
 		respondError(ctx, http.StatusInternalServerError, "invalid response", msg.ID)
 		return
@@ -144,7 +154,7 @@ func (r *Runtime) handleHTTPRequest(ctx *actor.Context, msg *proto.HTTPRequest)
 	ctx.Respond(resp)
 	r.stdout.Reset()
 
-	// only send metrics when its a request on LIVE
+	// only send metrics and logs when its a request on LIVE
 	if !msg.Preview {
 		metric := types.RequestMetric{
 			ID:           uuid.New(),
@@ -154,8 +164,14 @@ func (r *Runtime) handleHTTPRequest(ctx *actor.Context, msg *proto.HTTPRequest)
 			RequestURL: msg.URL,
 			StatusCode: status,
 		}
-		pid := ctx.Engine().Registry.GetPID(KindMetric, "1")
-		ctx.Send(pid, metric)
+		metricPID := ctx.Engine().Registry.GetPID(KindMetric, "1")
+		ctx.Send(metricPID, metric)
+
+		runtimeLogPID := ctx.Engine().Registry.GetPID(KindRuntimeLog, "1")
+		runtimeLog := types.RuntimeLogEvent{
+			Data: logs,
+		}
+		ctx.Send(runtimeLogPID, runtimeLog)
 	}
 }
 

internal/actrs/runtime_log.go

@@ -0,0 +1,23 @@
+package actrs
+
+import (
+	"github.com/anthdm/hollywood/actor"
+	"github.com/anthdm/raptor/internal/types"
+)
+
+const KindRuntimeLog = "runtime_log"
+
+type RuntimeLog struct{}
+
+func NewRuntimeLog() actor.Receiver {
+	return &RuntimeLog{}
+}
+
+func (rl *RuntimeLog) Receive(c *actor.Context) {
+	switch msg := c.Message().(type) {
+	case actor.Started:
+	case actor.Stopped:
+	case types.RuntimeLogEvent:
+		_ = msg
+	}
+}

internal/runtime/runtime.go

@@ -6,7 +6,6 @@ import (
 	"io"
 	"os"
 
-	"github.com/anthdm/raptor/internal/spidermonkey"
 	"github.com/google/uuid"
 	"github.com/tetratelabs/wazero"
 	"github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1"
@@ -41,14 +40,7 @@ func New(ctx context.Context, args Args) (*Runtime, error) {
 	}
 	wasi_snapshot_preview1.MustInstantiate(ctx, r.runtime)
 
-	switch args.Engine {
-	case "js":
-		r.blob = spidermonkey.WasmBlob
-	default:
-		r.blob = args.Blob
-	}
-
-	mod, err := r.runtime.CompileModule(ctx, r.blob)
+	mod, err := r.runtime.CompileModule(ctx, args.Blob)
 	if err != nil {
 		return nil, fmt.Errorf("runtime failed to compile module: %s", err)
 	}
@@ -57,10 +49,6 @@ func New(ctx context.Context, args Args) (*Runtime, error) {
 	return r, nil
 }
 
-func (r *Runtime) Blob() []byte {
-	return r.blob
-}
-
 func (r *Runtime) Invoke(stdin io.Reader, env map[string]string, args ...string) error {
 	modConf := wazero.NewModuleConfig().
 		WithStdin(stdin).

internal/runtime/runtime_test.go

@@ -8,14 +8,47 @@ import (
 	"testing"
 
 	"github.com/anthdm/raptor/internal/shared"
+	"github.com/anthdm/raptor/internal/spidermonkey"
 	"github.com/anthdm/raptor/proto"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"github.com/tetratelabs/wazero"
 	pb "google.golang.org/protobuf/proto"
 )
 
-func TestRuntime(t *testing.T) {
+func TestRuntimeInvokeJSCode(t *testing.T) {
+	b, err := os.ReadFile("../../examples/js/index.js")
+	require.Nil(t, err)
+
+	req := &proto.HTTPRequest{
+		Method: "get",
+		URL:    "/",
+		Body:   nil,
+	}
+	breq, err := pb.Marshal(req)
+	require.Nil(t, err)
+
+	out := &bytes.Buffer{}
+	args := Args{
+		Stdout:       out,
+		DeploymentID: uuid.New(),
+		Blob:         spidermonkey.WasmBlob,
+		Engine:       "js",
+		Cache:        wazero.NewCompilationCache(),
+	}
+	r, err := New(context.Background(), args)
+	require.Nil(t, err)
+
+	scriptArgs := []string{"", "-e", string(b)}
+	require.Nil(t, r.Invoke(bytes.NewReader(breq), nil, scriptArgs...))
+
+	_, _, status, err := shared.ParseStdout("js", out)
+	require.Nil(t, err)
+	require.Equal(t, http.StatusOK, status)
+	require.Nil(t, r.Close())
+}
+
+func TestRuntimeInvokeGoCode(t *testing.T) {
 	b, err := os.ReadFile("../../examples/go/app.wasm")
 	require.Nil(t, err)
 
@@ -38,7 +71,7 @@ func TestRuntime(t *testing.T) {
 	r, err := New(context.Background(), args)
 	require.Nil(t, err)
 	require.Nil(t, r.Invoke(bytes.NewReader(breq), nil))
-	_, status, err := shared.ParseRuntimeHTTPResponse(out.String())
+	_, _, status, err := shared.ParseStdout("go", out)
 	require.Nil(t, err)
 	require.Equal(t, http.StatusOK, status)
 	require.Nil(t, r.Close())

internal/shared/shared.go

@@ -13,8 +13,34 @@ import (
 	"github.com/anthdm/raptor/proto"
 )
 
+const magicLen = 8
+
 var errInvalidHTTPResponse = errors.New("invalid HTTP response")
 
+func ParseStdout(runtime string, stdout io.Reader) (logs []byte, resp []byte, status int, err error) {
+	stdoutb, err := io.ReadAll(GetDecodedStdout(runtime, stdout))
+	if err != nil {
+		return
+	}
+	outLen := len(stdoutb)
+	if outLen < magicLen {
+		err = fmt.Errorf("mallformed HTTP response missing last %d bytes", magicLen)
+		return
+	}
+	magicStart := outLen - magicLen
+	status = int(binary.LittleEndian.Uint32(stdoutb[magicStart : magicStart+4]))
+	respLen := binary.LittleEndian.Uint32(stdoutb[magicStart+4:])
+	fmt.Println(status)
+	if int(respLen) > outLen-magicLen {
+		err = fmt.Errorf("response length exceeds available data")
+		return
+	}
+	respStart := outLen - magicLen - int(respLen)
+	resp = stdoutb[respStart : respStart+int(respLen)]
+	logs = stdoutb[:respStart]
+	return
+}
+
 func ParseRuntimeHTTPResponse(in string) (resp string, status int, err error) {
 	if len(in) < 16 {
 		err = fmt.Errorf("misformed HTTP response missing last 16 bytes")
@@ -63,3 +89,14 @@ func makeProtoHeader(header http.Header) map[string]*proto.HeaderFields {
 	}
 	return m
 }
+
+func GetDecodedStdout(runtime string, stdout io.Reader) io.Reader {
+	switch runtime {
+	case "go":
+		return stdout
+	case "js":
+		return hex.NewDecoder(stdout)
+	default:
+		return stdout
+	}
+}