adds trigger data env and runtype to cmd output

Signed-off-by: Jordan Rash <15827604+jordan-rash@users.noreply.github.com>

api/nodecontrol/node-info-response.json

@@ -75,6 +75,10 @@
           "type": "string",
           "description": "The type of the workload"
         },
+        "workload_runtype": {
+          "type": "string",
+          "description": "The runtype/lifecycle of the workload"
+        },
         "workload_state": {
           "type": "string",
           "description": "The state of the workload"
@@ -86,6 +90,7 @@
         "start_time",
         "runtime",
         "workload_type",
+        "workload_runtype",
         "workload_state"
       ]
     }

api/nodecontrol/workload-ping-response.json

@@ -39,6 +39,10 @@
           "type": "string",
           "description": "The type of the workload"
         },
+        "workload_runtype": {
+          "type": "string",
+          "description": "The runtype/lifecycle of the workload"
+        },
         "workload_state": {
           "type": "string",
           "description": "The state of the workload"
@@ -50,6 +54,7 @@
         "start_time",
         "runtime",
         "workload_type",
+        "workload_runtype",
         "workload_state"
       ]
     }

cmd/nex/workload.go

@@ -17,6 +17,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"
 
 	"strings"
 
@@ -332,8 +333,21 @@ func (i InfoWorkload) Run(ctx context.Context, globals *Globals) error {
 	w := columns.New("Information about Workload %s", i.WorkloadId)
 	w.AddRow("Name", resp.WorkloadSummary.Name)
 	w.AddRow("Start Time", resp.WorkloadSummary.StartTime)
-	w.AddRowIf("Run Time", resp.WorkloadSummary.Runtime, resp.WorkloadSummary.Runtime != "")
+	w.AddRow("Run Time", func() string {
+		if resp.WorkloadSummary.Runtime == "" {
+			return "--"
+		}
+		if resp.WorkloadSummary.Runtime == "0s" {
+			t, err := time.Parse(time.DateTime, resp.WorkloadSummary.StartTime)
+			if err != nil {
+				return "--"
+			}
+			return time.Since(t).String()
+		}
+		return resp.WorkloadSummary.Runtime
+	}())
 	w.AddRow("Workload Type", resp.WorkloadSummary.WorkloadType)
+	w.AddRow("Workload Runtype", resp.WorkloadSummary.WorkloadRuntype)
 	w.AddRow("Workload State", resp.WorkloadSummary.WorkloadState)
 	s, err := w.Render()
 	if err != nil {

node/internal/actors/control_api_conversions.go

@@ -96,12 +96,13 @@ func infoResponseFromProto(response *actorproto.NodeInfo) *api.NodeInfoResponseJ
 
 	for _, workload := range response.Workloads {
 		ret.WorkloadSummaries = append(ret.WorkloadSummaries, api.WorkloadSummary{
-			Id:            workload.Id,
-			Name:          workload.Name,
-			Runtime:       workload.Runtime,
-			StartTime:     workload.StartedAt.AsTime().Format(time.DateTime),
-			WorkloadType:  workload.WorkloadType,
-			WorkloadState: workload.State,
+			Id:              workload.Id,
+			Name:            workload.Name,
+			Runtime:         workload.Runtime,
+			StartTime:       workload.StartedAt.AsTime().Format(time.DateTime),
+			WorkloadType:    workload.WorkloadType,
+			WorkloadRuntype: workload.WorkloadRuntype,
+			WorkloadState:   workload.State,
 		})
 	}
 	return ret
@@ -147,12 +148,13 @@ func pingResponseFromProto(response *actorproto.PingNodeResponse) *api.NodePingR
 func workloadPingResponseFromProto(response *actorproto.PingWorkloadResponse) *api.WorkloadPingResponseJson {
 	return &api.WorkloadPingResponseJson{
 		WorkloadSummary: &api.Workload{
-			Id:            response.Workload.Id,
-			Name:          response.Workload.Name,
-			Runtime:       response.Workload.Runtime,
-			StartTime:     response.Workload.StartedAt.AsTime().Format(time.DateTime),
-			WorkloadType:  response.Workload.WorkloadType,
-			WorkloadState: response.Workload.State,
+			Id:              response.Workload.Id,
+			Name:            response.Workload.Name,
+			Runtime:         response.Workload.Runtime,
+			StartTime:       response.Workload.StartedAt.AsTime().Format(time.DateTime),
+			WorkloadType:    response.Workload.WorkloadType,
+			WorkloadRuntype: response.Workload.WorkloadRuntype,
+			WorkloadState:   response.Workload.State,
 		},
 	}
 }

node/internal/actors/pb/agents.proto

@@ -159,8 +159,9 @@ message WorkloadSummary {
     string runtime = 3;
     google.protobuf.Timestamp started_at = 4;
     string workload_type = 5;
-    string namespace = 6;
-    string state = 7;
+    string workload_runtype = 6;
+    string namespace = 7;
+    string state = 8;
 }
 
 message SetLameDuck {}

node/internal/actors/process_actor.go

@@ -2,6 +2,7 @@ package actors
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"time"
 
@@ -107,13 +108,14 @@ func (a *processActor) Receive(ctx *goakt.ReceiveContext) {
 		ctx.Shutdown()
 	case *actorproto.QueryWorkload:
 		ctx.Response(&actorproto.WorkloadSummary{
-			Id:           a.id,
-			Name:         a.processName,
-			Namespace:    a.namespace,
-			Runtime:      a.runTime.String(),
-			StartedAt:    timestamppb.New(a.startedAt),
-			WorkloadType: a.workloadType,
-			State:        a.state,
+			Id:              a.id,
+			Name:            a.processName,
+			Namespace:       a.namespace,
+			Runtime:         a.runTime.String(),
+			StartedAt:       timestamppb.New(a.startedAt),
+			WorkloadType:    a.workloadType,
+			WorkloadRuntype: a.runType,
+			State:           a.state,
 		})
 	default:
 		a.logger.Warn("unknown message", slog.Any("msg", ctx.Message()))
@@ -158,7 +160,8 @@ func (a *processActor) SpawnOsProcess(ctx *goakt.ReceiveContext) {
 		a.cancel = cancel
 
 		// TODO: subscribe to all triggers or change to only allow one
-		s, err := a.nc.Subscribe(a.triggerSubs[0], func(msg *nats.Msg) {
+		// TODO: need to have a better quere group. possibly an ID at start time
+		s, err := a.nc.QueueSubscribe(a.triggerSubs[0], a.processName, func(msg *nats.Msg) {
 			a.state = models.WorkloadStateRunning
 
 			ticker := time.NewTicker(DefaultJobRunTime)
@@ -175,6 +178,10 @@ func (a *processActor) SpawnOsProcess(ctx *goakt.ReceiveContext) {
 			stdout := logCapture{logger: a.logger, nc: a.nc, namespace: a.namespace, name: a.id, stderr: false}
 			stderr := logCapture{logger: a.logger, nc: a.nc, namespace: a.namespace, name: a.id, stderr: true}
 
+			if a.env == nil {
+				a.env = make(map[string]string)
+			}
+			a.env["NEX_TRIGGER_DATA"] = string(msg.Data)
 			a.process, err = NewOsProcess(a.id, a.ref.LocalCachePath, a.env, a.argv, a.logger, stdout, stderr)
 			if err != nil {
 				a.logger.Error("failed to create process", slog.Any("err", err))
@@ -192,6 +199,7 @@ func (a *processActor) SpawnOsProcess(ctx *goakt.ReceiveContext) {
 
 			a.runTime = a.runTime + exeEnd.Sub(exeStart)
 			a.state = models.WorkloadStateWarm
+			ticker.Stop()
 		})
 		if err != nil {
 			a.logger.Error("failed to subscribe to trigger", slog.Any("err", err))

test/testdata/function/main.go

@@ -8,6 +8,7 @@ import (
 
 func main() {
 	fmt.Fprint(os.Stdout, "starting function")
+	fmt.Fprint(os.Stdout, os.Getenv("NEX_TRIGGER_DATA"))
 	time.Sleep(750 * time.Millisecond)
 	fmt.Fprint(os.Stdout, "ending function")
 }

test/workload_test.go

@@ -271,6 +271,7 @@ func TestDirectStartFunction(t *testing.T) {
 	}()
 
 	passed := false
+	triggerLogs := new(bytes.Buffer)
 	go func() {
 		pub, err := kp.PublicKey()
 		if err != nil {
@@ -341,8 +342,21 @@ func TestDirectStartFunction(t *testing.T) {
 			t.Error(err)
 			return
 		}
+		defer nc.Close()
 
-		err = nc.Publish("test", []byte("test"))
+		workloadId := match[1]
+		sub, err := nc.Subscribe("$NEX.logs.system."+workloadId+".stdout", func(msg *nats.Msg) {
+			triggerLogs.Write(msg.Data)
+		})
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		defer func() {
+			_ = sub.Drain()
+		}()
+
+		err = nc.Publish("test", []byte("test data 123"))
 		if err != nil {
 			t.Error(err)
 			return
@@ -392,4 +406,8 @@ func TestDirectStartFunction(t *testing.T) {
 	if !passed {
 		t.Fatal("expected workload to run for 500ms - 1s")
 	}
+
+	if !bytes.Contains(triggerLogs.Bytes(), []byte("test data 123")) {
+		t.Error("expected 'test data 123' to be consumed by workload")
+	}
 }