Distributed runtimes

Makefile

@@ -1,12 +1,18 @@
 .PHONY: proto
 
+PROTO_PATH := "../../go/pkg/mod/github.com/anthdm/[email protected]/actor"
+
 build:
 	@go build -o bin/api cmd/api/main.go 
-	@go build -o bin/wasmserver cmd/wasmserver/main.go 
+	@go build -o bin/ingress cmd/ingress/main.go 
 	@go build -o bin/raptor cmd/cli/main.go 
+	@go build -o bin/runtime cmd/runtime/main.go 
+
+ingress: build
+	@./bin/ingress
 
-wasmserver: build
-	@./bin/wasmserver
+runtime: build
+	@./bin/runtime
 
 api: build
 	@./bin/api --seed
@@ -15,7 +21,7 @@ test:
 	@go test ./internal/* -v
 
 proto:
-	protoc --go_out=. --go_opt=paths=source_relative --proto_path=. proto/types.proto
+	protoc --go_out=. --go_opt=paths=source_relative --proto_path=$(PROTO_PATH) --proto_path=. proto/types.proto
 
 clean:
 	@rm -rf bin/api

cmd/api/main.go

@@ -50,8 +50,8 @@ func main() {
 	}
 
 	server := api.NewServer(store, store, modCache)
-	fmt.Printf("api server running\t%s\n", config.GetApiUrl())
-	log.Fatal(server.Listen(config.Get().APIServerAddr))
+	fmt.Printf("api server running\t%s\n", config.ApiUrl())
+	log.Fatal(server.Listen(config.Get().HTTPAPIAddr))
 }
 
 func seedEndpoint(store storage.Store, cache storage.ModCacher) {
@@ -81,7 +81,7 @@ func seedEndpoint(store storage.Store, cache storage.ModCacher) {
 	if err != nil {
 		log.Fatal(err)
 	}
-	fmt.Printf("endpoint seeded: %s/live/%s\n", config.GetWasmUrl(), endpoint.ID)
+	fmt.Printf("endpoint seeded: %s/live/%s\n", config.IngressUrl(), endpoint.ID)
 }
 
 func compile(ctx context.Context, cache wazero.CompilationCache, blob []byte) {

cmd/cli/main.go

@@ -61,7 +61,7 @@ func main() {
 		printUsage()
 	}
 
-	c := client.New(client.NewConfig().WithURL(config.GetApiUrl()))
+	c := client.New(client.NewConfig().WithURL(config.ApiUrl()))
 	command := command{
 		client: c,
 	}
@@ -175,7 +175,7 @@ func (c command) handleDeploy(args []string) {
 	}
 	fmt.Println(string(b))
 	fmt.Println()
-	fmt.Printf("deploy preview: %s/preview/%s\n", config.GetWasmUrl(), deploy.ID)
+	fmt.Printf("deploy preview: %s/preview/%s\n", config.IngressUrl(), deploy.ID)
 }
 
 func (c command) handleServeEndpoint(args []string) {

cmd/wasmserver/main.go → cmd/ingress/main.go

@@ -16,13 +16,21 @@ import (
 )
 
 func main() {
-	var configFile string
-	flagSet := flag.NewFlagSet("raptor", flag.ExitOnError)
+	var (
+		configFile string
+		address    string
+		id         string
+		region     string
+	)
+
+	flagSet := flag.NewFlagSet("ingress", flag.ExitOnError)
 	flagSet.StringVar(&configFile, "config", "config.toml", "")
+	flagSet.StringVar(&address, "cluster-addr", "127.0.0.1:8132", "")
+	flagSet.StringVar(&id, "id", "ingress", "")
+	flagSet.StringVar(&region, "region", "default", "")
 	flagSet.Parse(os.Args[1:])
 
-	err := config.Parse(configFile)
-	if err != nil {
+	if err := config.Parse(configFile); err != nil {
 		log.Fatal(err)
 	}
 
@@ -44,27 +52,27 @@ func main() {
 	)
 
 	clusterConfig := cluster.NewConfig().
-		WithListenAddr(config.Get().Cluster.Address).
-		WithRegion(config.Get().Cluster.Region).
-		WithID(config.Get().Cluster.ID)
+		WithListenAddr(address).
+		WithRegion(region).
+		WithID(id)
 	c, err := cluster.New(clusterConfig)
 	if err != nil {
 		log.Fatal(err)
 	}
 	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.Spawn(actrs.NewRuntimeManager(c), actrs.KindRuntimeManager, actor.WithID("1"))
 	c.Engine().Spawn(actrs.NewRuntimeLog, actrs.KindRuntimeLog, actor.WithID("1"))
 	c.Start()
 
 	server := actrs.NewWasmServer(
-		config.Get().WASMServerAddr,
+		config.Get().HTTPIngressAddr,
 		c,
 		store,
 		metricStore,
 		modCache)
 	c.Engine().Spawn(server, actrs.KindWasmServer)
-	fmt.Printf("wasm server running\t%s\n", config.Get().WASMServerAddr)
+	fmt.Printf("ingress server running\t%s\n", config.Get().HTTPIngressAddr)
 
 	sigch := make(chan os.Signal, 1)
 	signal.Notify(sigch, syscall.SIGINT, syscall.SIGTERM)

cmd/runtime/main.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/anthdm/hollywood/actor"
+	"github.com/anthdm/hollywood/cluster"
+	"github.com/anthdm/raptor/internal/actrs"
+	"github.com/anthdm/raptor/internal/config"
+	"github.com/anthdm/raptor/internal/storage"
+)
+
+func main() {
+	var (
+		configFile string
+		address    string
+		id         string
+		region     string
+	)
+
+	flagSet := flag.NewFlagSet("runtime", flag.ExitOnError)
+	flagSet.StringVar(&configFile, "config", "config.toml", "")
+	flagSet.StringVar(&address, "cluster-addr", "127.0.0.1:8134", "")
+	flagSet.StringVar(&id, "id", "runtime", "")
+	flagSet.StringVar(&region, "region", "default", "")
+	flagSet.Parse(os.Args[1:])
+
+	if err := config.Parse(configFile); err != nil {
+		log.Fatal(err)
+	}
+
+	var (
+		user    = config.Get().Storage.User
+		pw      = config.Get().Storage.Password
+		dbname  = config.Get().Storage.Name
+		host    = config.Get().Storage.Host
+		port    = config.Get().Storage.Port
+		sslmode = config.Get().Storage.SSLMode
+	)
+	store, err := storage.NewSQLStore(user, pw, dbname, host, port, sslmode)
+	if err != nil {
+		log.Fatal(err)
+	}
+	var (
+		modCache = storage.NewDefaultModCache()
+		// metricStore = store
+	)
+	clusterConfig := cluster.NewConfig().
+		WithListenAddr(address).
+		WithRegion(region).
+		WithID(id)
+	c, err := cluster.New(clusterConfig)
+	if err != nil {
+		log.Fatal(err)
+	}
+	c.RegisterKind(actrs.KindRuntime, actrs.NewRuntime(store, modCache), &cluster.KindConfig{})
+	c.Engine().Spawn(actrs.NewMetric, actrs.KindMetric, actor.WithID("1"))
+	c.Engine().Spawn(actrs.NewRuntimeLog, actrs.KindRuntimeLog, actor.WithID("1"))
+	c.Start()
+
+	sigch := make(chan os.Signal, 1)
+	signal.Notify(sigch, syscall.SIGINT, syscall.SIGTERM)
+	<-sigch
+}

cmd/runtime/runtime.md

@@ -0,0 +1 @@
+# Runtime

examples/go/main.go

@@ -1,15 +1,15 @@
 package main
 
 import (
-	"fmt"
 	"net/http"
 
 	raptor "github.com/anthdm/raptor/sdk"
+	"github.com/go-chi/chi"
 )
 
 func handleLogin(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
-	w.Write([]byte("hello from the login handler"))
+	w.Write([]byte("hello from the login handler YADA"))
 }
 
 func handleDashboard(w http.ResponseWriter, r *http.Request) {
@@ -18,9 +18,8 @@ func handleDashboard(w http.ResponseWriter, r *http.Request) {
 }
 
 func main() {
-	// router := chi.NewMux()
-	// router.Get("/dashboard", handleDashboard)
-	// router.Get("/login", handleLogin)
-	fmt.Println("user log")
+	router := chi.NewMux()
+	router.Get("/dashboard", handleDashboard)
+	router.Get("/login", handleLogin)
 	raptor.Handle(http.HandlerFunc(handleLogin))
 }

go.mod

@@ -35,6 +35,7 @@ require (
 )
 
 require (
+	github.com/go-chi/chi v1.5.5
 	github.com/pelletier/go-toml/v2 v2.1.1
 	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/sys v0.14.0 // indirect

go.sum

@@ -7,6 +7,8 @@ github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QH
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-chi/chi v1.5.5 h1:vOB/HbEMt9QqBqErz07QehcOKHaWFtuj87tTDVz2qXE=
+github.com/go-chi/chi v1.5.5/go.mod h1:C9JqLr3tIYjDOZpzn+BCuxY8z8vmca43EeMgyZt7irw=
 github.com/go-chi/chi/v5 v5.0.11 h1:BnpYbFZ3T3S1WMpD79r7R5ThWX40TaFB7L31Y8xqSwA=
 github.com/go-chi/chi/v5 v5.0.11/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=

internal/actrs/runtime.go

@@ -58,20 +58,25 @@ func (r *Runtime) Receive(c *actor.Context) {
 	case actor.Started:
 		r.started = time.Now()
 		r.repeat = c.SendRepeat(c.PID(), shutdown{}, runtimeKeepAlive)
-		r.managerPID = c.Engine().Registry.GetPID(KindRuntimeManager, "1")
 	case actor.Stopped:
+		r.repeat.Stop()
 		// TODO: send metrics about the runtime to the metric actor.
 		_ = time.Since(r.started)
-		c.Send(r.managerPID, removeRuntime{key: r.deploymentID.String()})
+		c.Send(r.managerPID, &proto.RemoveRuntime{Key: r.deploymentID.String()})
 		r.runtime.Close()
 		// Releasing this mod will invalidate the cache for some reason.
 		// r.mod.Close(context.TODO())
 	case *proto.HTTPRequest:
+		slog.Info("runtime handling request", "request_id", msg.ID, "pid", c.PID())
 		// Refresh the keepAlive timer
 		r.repeat = c.SendRepeat(c.PID(), shutdown{}, runtimeKeepAlive)
 		if r.runtime == nil {
 			r.initialize(msg)
 		}
+		// In the ideal world we should ask the cluster for the PID of the manager we
+		// need to notify we are done invoking. Hollywood does not have that functionality
+		// yet. To fix this we have the PID of the manager in the request messsage.
+		r.managerPID = msg.ManagerPID
 		// Handle the HTTP request that is forwarded from the WASM server actor.
 		r.handleHTTPRequest(c, msg)
 	case shutdown:

internal/actrs/runtime_manager.go

@@ -3,6 +3,7 @@ package actrs
 import (
 	"github.com/anthdm/hollywood/actor"
 	"github.com/anthdm/hollywood/cluster"
+	"github.com/anthdm/raptor/proto"
 )
 
 const KindRuntimeManager = "runtime_manager"
@@ -11,15 +12,10 @@ type (
 	requestRuntime struct {
 		key string
 	}
-	addRuntime struct {
-		key string
-		pid *actor.PID
-	}
-	removeRuntime struct {
-		key string
-	}
 )
 
+// RuntimeManager is an actor/receiver that is responsible for managing
+// runtimes across the cluster.
 type RuntimeManager struct {
 	runtimes map[string]*actor.PID
 	cluster  *cluster.Cluster
@@ -43,10 +39,8 @@ func (rm *RuntimeManager) Receive(c *actor.Context) {
 			rm.runtimes[msg.key] = pid
 		}
 		c.Respond(pid)
-	case addRuntime:
-		rm.runtimes[msg.key] = msg.pid
-	case removeRuntime:
-		delete(rm.runtimes, msg.key)
+	case *proto.RemoveRuntime:
+		delete(rm.runtimes, msg.Key)
 	case actor.Started:
 	case actor.Stopped:
 	case actor.Initialized:

internal/actrs/wasmserver.go

@@ -31,26 +31,26 @@ func newRequestWithResponse(request *proto.HTTPRequest) requestWithResponse {
 
 // WasmServer is an HTTP server that will proxy and route the request to the corresponding function.
 type WasmServer struct {
-	server         *http.Server
-	self           *actor.PID
-	store          storage.Store
-	metricStore    storage.MetricStore
-	cache          storage.ModCacher
-	cluster        *cluster.Cluster
-	responses      map[string]chan *proto.HTTPResponse
-	runtimeManager *actor.PID
+	server            *http.Server
+	self              *actor.PID
+	store             storage.Store
+	metricStore       storage.MetricStore
+	cache             storage.ModCacher
+	cluster           *cluster.Cluster
+	responses         map[string]chan *proto.HTTPResponse
+	runtimeManagerPID *actor.PID
 }
 
 // NewWasmServer return a new wasm server given a storage and a mod cache.
 func NewWasmServer(addr string, cluster *cluster.Cluster, store storage.Store, metricStore storage.MetricStore, cache storage.ModCacher) actor.Producer {
 	return func() actor.Receiver {
 		s := &WasmServer{
-			store:          store,
-			metricStore:    metricStore,
-			cache:          cache,
-			cluster:        cluster,
-			responses:      make(map[string]chan *proto.HTTPResponse),
-			runtimeManager: cluster.Engine().Registry.GetPID(KindRuntimeManager, "1"),
+			store:             store,
+			metricStore:       metricStore,
+			cache:             cache,
+			cluster:           cluster,
+			responses:         make(map[string]chan *proto.HTTPResponse),
+			runtimeManagerPID: cluster.Engine().Registry.GetPID(KindRuntimeManager, "1"),
 		}
 		server := &http.Server{
 			Handler: s,
@@ -67,14 +67,15 @@ func (s *WasmServer) Receive(c *actor.Context) {
 		s.initialize(c)
 	case actor.Stopped:
 	case requestWithResponse:
-		s.responses[msg.request.ID] = msg.response
 		// TODO: let's say the manager is not able to respond in time for some reason
 		// I think we might need to spawn a new runtime right here.
 		pid := s.requestRuntime(c, msg.request.DeploymentID)
 		if pid == nil {
 			slog.Error("failed to request a runtime PID")
 			return
 		}
+		s.responses[msg.request.ID] = msg.response
+		msg.request.ManagerPID = s.runtimeManagerPID
 		s.cluster.Engine().SendWithSender(pid, msg.request, s.self)
 	case *proto.HTTPResponse:
 		if resp, ok := s.responses[msg.RequestID]; ok {
@@ -93,9 +94,9 @@ func (s *WasmServer) initialize(c *actor.Context) {
 
 // NOTE: There could be a case where we do not get a response in time, hence
 // the PID will be nil. This case is handled where we should spawn the runtime
-// ourselfs.
+// ourselves.
 func (s *WasmServer) requestRuntime(c *actor.Context, key string) *actor.PID {
-	res, err := c.Request(s.runtimeManager, requestRuntime{
+	res, err := c.Request(s.runtimeManagerPID, requestRuntime{
 		key: key,
 	}, time.Millisecond*5).Result()
 	if err != nil {

internal/api/server.go

@@ -194,7 +194,7 @@ func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) error {
 
 	resp := PublishResponse{
 		DeploymentID: deploy.ID,
-		URL:          fmt.Sprintf("%s/live/%s", config.GetWasmUrl(), endpoint.ID),
+		URL:          fmt.Sprintf("%s/live/%s", config.IngressUrl(), endpoint.ID),
 	}
 	return writeJSON(w, http.StatusOK, resp)
 }