Direct Start Workload

agents/common/nexagent.go

@@ -68,9 +68,9 @@ func (agent *NexAgent) Run() error {
 
 	switch os.Args[1] {
 	case "up":
-		agent.callback.Up()
+		return agent.callback.Up()
 	case "preflight":
-		agent.callback.Preflight()
+		return agent.callback.Preflight()
 	default:
 		fmt.Println("No command supplied")
 	}
@@ -138,10 +138,12 @@ func (agent *NexAgent) handleStartWorkload(m *nats.Msg) {
 	var req agentapigen.StartWorkloadRequestJson
 	err := json.Unmarshal(m.Data, &req)
 	if err != nil {
+		_ = 0 // for linter
 		// TODO: return error envelope
 	}
 	err = agent.callback.StartWorkload(&req)
 	if err != nil {
+		_ = 0 // for linter
 		// TODO: return error envelope
 	}
 	// TODO: return success envelope
@@ -151,17 +153,19 @@ func (agent *NexAgent) handleStopWorkload(m *nats.Msg) {
 	var req agentapigen.StopWorkloadRequestJson
 	err := json.Unmarshal(m.Data, &req)
 	if err != nil {
+		_ = 0 // for linter
 		// TODO: return error envelope
 	}
 	err = agent.callback.StopWorkload(&req)
 	if err != nil {
+		_ = 0 // for linter
 		// TODO: return error envelope
 	}
 	// TODO: return success envelope
 }
 
 func (agent *NexAgent) handleListWorkloads(m *nats.Msg) {
-	agent.callback.ListWorkloads()
+	_ = agent.callback.ListWorkloads()
 }
 
 // TODO: this should actually get passed the node configuration options to make

agents/tester/go.mod

@@ -2,19 +2,15 @@ module github.com/synadia-io/nex/agents/tester
 
 go 1.23.2
 
-replace github.com/synadia-io/nex/agents/common => ../common
+replace github.com/synadia-io/nex => ../..
 
-replace github.com/synadia-io/nex/api/agent/go => ../../api/agent/go
-
-require github.com/synadia-io/nex/agents/common v0.0.0-00010101000000-000000000000
-
-require github.com/synadia-io/nex/api/agent/go v0.0.0-00010101000000-000000000000
+require github.com/synadia-io/nex v0.0.0-00010101000000-000000000000
 
 require (
-	github.com/klauspost/compress v1.17.2 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/nats-io/nats.go v1.37.0 // indirect
 	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/crypto v0.28.0 // indirect
+	golang.org/x/sys v0.26.0 // indirect
 )

agents/tester/go.sum

@@ -1,12 +1,12 @@
-github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
-github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE=
 github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
 github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
-golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
-golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
-golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw=
+golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U=
+golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=
+golang.org/x/sys v0.26.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

api/agent/go/subjects.go

@@ -16,9 +16,9 @@ func StartWorkloadSubscribeSubject(workloadType string) string {
 }
 
 func StopWorkloadSubscribeSubject(workloadType string) string {
-	return fmt.Sprintf("agent.%s.workloads.stop")
+	return fmt.Sprintf("agent.%s.workloads.stop", workloadType)
 }
 
 func ListWorkloadsSubscribeSubject(workloadType string) string {
-	return fmt.Sprintf("agent.%s.workloads.list")
+	return fmt.Sprintf("agent.%s.workloads.list", workloadType)
 }

api/nodecontrol/client.go

@@ -113,13 +113,13 @@ func (c *ControlAPIClient) DeployWorkload(namespace, nodeId string, req nodegen.
 		return nil, err
 	}
 
-	resp := new(nodegen.StartWorkloadResponseJson)
-	err = json.Unmarshal(msg.Data, resp)
+	outerEnv := new(actors.Envelope[nodegen.StartWorkloadResponseJson])
+	err = json.Unmarshal(msg.Data, outerEnv)
 	if err != nil {
 		return nil, err
 	}
 
-	return resp, nil
+	return &outerEnv.Data, nil
 }
 
 func (c *ControlAPIClient) UndeployWorkload(nodeId, workloadId string, req nodegen.StopWorkloadRequestJson) (*nodegen.StopWorkloadResponseJson, error) {
@@ -133,13 +133,13 @@ func (c *ControlAPIClient) UndeployWorkload(nodeId, workloadId string, req nodeg
 		return nil, err
 	}
 
-	resp := new(nodegen.StopWorkloadResponseJson)
-	err = json.Unmarshal(msg.Data, resp)
+	outerEnv := new(actors.Envelope[nodegen.StopWorkloadResponseJson])
+	err = json.Unmarshal(msg.Data, outerEnv)
 	if err != nil {
 		return nil, err
 	}
 
-	return resp, nil
+	return &outerEnv.Data, nil
 }
 
 func (c *ControlAPIClient) GetInfo(nodeId, namespace string) (*nodegen.NodeInfoResponseJson, error) {

cmd/nex/cli_unix.go

@@ -14,7 +14,7 @@ type Globals struct {
 
 	Config              kong.ConfigFlag  `help:"Configuration file to load" placeholder:"./nex.config.json"`
 	Version             kong.VersionFlag `help:"Print version information"`
-	Namespace           string           `env:"NEX_NAMESPACE" placeholder:"default" help:"Specifies namespace when running nex commands"`
+	Namespace           string           `env:"NEX_NAMESPACE" default:"default" help:"Specifies namespace when running nex commands"`
 	Check               bool             `help:"Print the current values of all options without running a command"`
 	DisableUpgradeCheck bool             `env:"NEX_DISABLE_UPGRADE_CHECK" name:"disable-upgrade-check" help:"Disable the upgrade check"`
 	AutoUpgrade         bool             `env:"NEX_AUTO_UPGRADE" name:"auto-upgrade" help:"Automatically upgrade the nex CLI when a new version is available"`

cmd/nex/cli_windows.go

@@ -10,7 +10,7 @@ type Globals struct {
 
 	Config    kong.ConfigFlag  `help:"Configuration file to load" placeholder:"./nex.config.json"`
 	Version   kong.VersionFlag `help:"Print version information"`
-	Namespace string           `env:"NEX_NAMESPACE" placeholder:"default" help:"Specifies namespace when running nex commands"`
+	Namespace string           `env:"NEX_NAMESPACE" default:"default" help:"Specifies namespace when running nex commands"`
 	Check     bool             `help:"Print the current values of all options without running a command"`
 }
 

cmd/nex/workload.go

@@ -42,7 +42,7 @@ type RunWorkload struct {
 	WorkloadPublicKey        string            `name:"public-key" description:"Public key of the workload"`
 	WorkloadUri              string            `name:"uri" description:"URI of the workload.  file:// oci:// nats://" placeholder:"file://./workload"`
 	WorkloadTriggerSubjects  []string          `name:"triggers" description:"Subjects to trigger the workload"`
-	WorkloadType             string            `name:"type" description:"Type of workload" default:"direct"`
+	WorkloadType             string            `name:"type" description:"Type of workload" default:"direct_start"`
 }
 
 func (RunWorkload) AfterApply(globals *Globals) error {
@@ -115,6 +115,7 @@ func (r RunWorkload) Run(ctx context.Context, globals *Globals, w *Workload) err
 		TriggerSubjects: r.WorkloadTriggerSubjects,
 		Uri:             r.WorkloadUri,
 		WorkloadName:    r.WorkloadName,
+		WorkloadType:    r.WorkloadType,
 	}
 
 	resp, err := controller.DeployWorkload(globals.Namespace, r.NodeId, startRequest)
@@ -123,7 +124,7 @@ func (r RunWorkload) Run(ctx context.Context, globals *Globals, w *Workload) err
 	}
 
 	if resp.Started {
-		fmt.Printf("Workload %s[%s] started on node %s\n", r.WorkloadName, resp.Id, r.NodeId)
+		fmt.Printf("Workload %s [%s] started on node %s\n", r.WorkloadName, resp.Id, r.NodeId)
 	} else {
 		fmt.Printf("Workload %s failed to start on node %s\n", r.WorkloadName, r.NodeId)
 	}
@@ -132,8 +133,9 @@ func (r RunWorkload) Run(ctx context.Context, globals *Globals, w *Workload) err
 }
 
 type StopWorkload struct {
-	NodeId     string `description:"Node ID of the node the workload in running"`
-	WorkloadId string `description:"ID of the workload to stop" required:"true"`
+	NodeId       string `description:"Node ID of the node the workload in running"`
+	WorkloadId   string `description:"ID of the workload to stop" required:"true"`
+	WorkloadType string `name:"type" description:"Type of workload" default:"direct_start"`
 }
 
 func (StopWorkload) AfterApply(globals *Globals) error {
@@ -161,8 +163,9 @@ func (s StopWorkload) Run(ctx context.Context, globals *Globals, w *Workload) er
 	}
 
 	req := gen.StopWorkloadRequestJson{
-		NodeId:     s.NodeId,
-		WorkloadId: s.WorkloadId,
+		NodeId:       s.NodeId,
+		WorkloadId:   s.WorkloadId,
+		WorkloadType: s.WorkloadType,
 	}
 
 	resp, err := controller.UndeployWorkload(s.NodeId, globals.Namespace, req)

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/nats-io/nats-server/v2 v2.10.22
 	github.com/nats-io/nats.go v1.37.0
 	github.com/nats-io/nkeys v0.4.7
+	github.com/nats-io/nuid v1.0.1
 	github.com/stretchr/testify v1.9.0
 	github.com/tochemey/goakt/v2 v2.7.1
 	google.golang.org/protobuf v1.35.1
@@ -46,7 +47,6 @@ require (
 	github.com/mschoch/smat v0.2.0 // indirect
 	github.com/muesli/termenv v0.15.2 // indirect
 	github.com/nats-io/jwt/v2 v2.7.2 // indirect
-	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/redis/go-redis/v9 v9.6.1 // indirect

node/actors/agent_supervisor.go

@@ -43,7 +43,6 @@ func (s *AgentSupervisor) Receive(ctx *goakt.ReceiveContext) {
 	switch ctx.Message().(type) {
 	case *goaktpb.PostStart:
 		s.logger = ctx.Self().Logger()
-		s.logger.Infof("Agent supervisor '%v' is running", ctx.Self().Name())
 	case *actorproto.QueryWorkloads:
 		s.queryWorkloads(ctx)
 	default:

node/artifacts.go → node/actors/artifacts.go

@@ -1,6 +1,10 @@
-package node
+package actors
 
-import "net/url"
+import (
+	"net/url"
+	"os"
+	"strings"
+)
 
 const (
 	SchemeFile = "file"
@@ -24,24 +28,57 @@ type ArtifactReference struct {
 	Size int
 }
 
+func extractTag(location *url.URL) string {
+	return location.Fragment
+}
+
 // Obtains an artifact from the specified location. If the indicated location allows for
 // differentiation using tags (e.g. OCI, Object Store), then the supplied tag will be used,
 // otherwise it will be ignored
-func GetArtifact(name string, tag string, location *url.URL) (*ArtifactReference, error) {
+func getArtifact(name string, uri string) (*ArtifactReference, error) {
+	location, err := url.Parse(uri)
+	if err != nil {
+		return nil, err
+	}
+
 	switch location.Scheme {
 	case SchemeFile:
 		return cacheFile(name, location)
 	case SchemeNATS:
-		return cacheObjectStoreArtifact(name, tag, location)
+		return cacheObjectStoreArtifact(name, extractTag(location), location)
 	case SchemeOCI:
-		return cacheOciArtifact(name, tag, location)
+		return cacheOciArtifact(name, extractTag(location), location)
 	}
 	return nil, nil
 }
 
 func cacheFile(name string, location *url.URL) (*ArtifactReference, error) {
-	// TODO: Implement
-	return nil, nil
+	filePath, tag := "", ""
+	sPath := strings.SplitN(location.Path, ":", 2) // splits on first : only
+	if len(sPath) == 2 {
+		filePath = sPath[0]
+		tag = sPath[1]
+	} else {
+		filePath = location.Path
+	}
+
+	info, err := os.Stat(filePath)
+	if err != nil {
+		return nil, err
+	}
+
+	if info.IsDir() {
+		return nil, nil
+	}
+
+	return &ArtifactReference{
+		Name:             name,
+		Tag:              tag,
+		OriginalLocation: location,
+		LocalCachePath:   filePath,
+		Digest:           "",
+		Size:             int(info.Size()),
+	}, nil
 }
 
 func cacheObjectStoreArtifact(name string, tag string, location *url.URL) (*ArtifactReference, error) {