Merge branch 'main' into test/actors

synadia-io · Jan 3, 2025 · bffa94a · bffa94a
2 parents 5dd61eb + c7a4459
commit bffa94a
Show file tree

Hide file tree

Showing 14 changed files with 83 additions and 21 deletions.
diff --git a/api/nodecontrol/client.go b/api/nodecontrol/client.go
@@ -11,9 +11,7 @@ import (
 	"github.com/synadia-io/nex/models"
 )
 
-var (
-	DefaultRequestTimeout = 5 * time.Second
-)
+var DefaultRequestTimeout = 5 * time.Second
 
 type ControlAPIClient struct {
 	nc     *nats.Conn

diff --git a/api/nodecontrol/gen/start_workload_response.go b/api/nodecontrol/gen/start_workload_response.go
diff --git a/api/nodecontrol/gen/stop_workload_response.go b/api/nodecontrol/gen/stop_workload_response.go
diff --git a/api/nodecontrol/start-workload-response.json b/api/nodecontrol/start-workload-response.json
@@ -15,13 +15,17 @@
     },
     "started": {
       "type": "boolean"
+    },
+    "message": {
+      "type": "string"
     }
   },
   "required": [
     "id",
     "issuer",
     "name",
-    "started"
+    "started",
+    "message"
   ],
   "definitions": {},
   "additionalProperties": false

diff --git a/api/nodecontrol/stop-workload-response.json b/api/nodecontrol/stop-workload-response.json
@@ -12,12 +12,16 @@
     },
     "stopped": {
       "type": "boolean"
+    },
+    "message": {
+      "type": "string"
     }
   },
   "required": [
     "id",
     "issuer",
-    "stopped"
+    "stopped",
+    "message"
   ],
   "definitions": {},
   "additionalProperties": false

diff --git a/cmd/nex/logger.go b/cmd/nex/logger.go
@@ -10,6 +10,7 @@ import (
 
 	"disorder.dev/shandler"
 	"github.com/nats-io/nats.go"
+	"github.com/synadia-io/nex/models"
 )
 
 func configureLogger(cfg *Globals, nc *nats.Conn, serverPublicKey string, showSystemLogs, hideWorkloadLogs bool) *slog.Logger {
@@ -85,8 +86,8 @@ func configureLogger(cfg *Globals, nc *nats.Conn, serverPublicKey string, showSy
 		}
 	}
 	if slices.Contains(cfg.Target, "nats") {
-		natsLogSubject := fmt.Sprintf("$NEX.logs.%s.stdout", serverPublicKey)
-		natsErrLogSubject := fmt.Sprintf("$NEX.logs.%s.stderr", serverPublicKey)
+		natsLogSubject := fmt.Sprintf("%s.%s.%s.stdout", models.LogAPIPrefix, models.NodeSystemNamespace, serverPublicKey)
+		natsErrLogSubject := fmt.Sprintf("%s.%s.%s.stderr", models.LogAPIPrefix, models.NodeSystemNamespace, serverPublicKey)
 		stdoutWriters = append(stdoutWriters, NewNatsLogger(nc, natsLogSubject))
 		stderrWriters = append(stderrWriters, NewNatsLogger(nc, natsErrLogSubject))
 	}

diff --git a/cmd/nex/node.go b/cmd/nex/node.go
@@ -348,6 +348,8 @@ type Up struct {
 	HideWorkloadLogs                 bool              `name:"hide-workload-logs" help:"Hide logs from workloads" default:"false"`
 	ShowSystemLogs                   bool              `name:"system-logs" help:"Show verbose level logs from inside actor framework" default:"false" hidden:""`
 	OCICache                         string            `name:"oci-cache" help:"Path to OCI cache registry" placeholder:"localhost:5000"`
+	StartMessage                     string            `name:"start-message" help:"Message to display on the successful start of a workload" placeholder:"Great job!"`
+	StopMessage                      string            `name:"stop-message" help:"Message to display on the successful stop of a workload" placeholder:"Goodbye!"`
 
 	HostServicesConfig HostServicesConfig `embed:"" prefix:"hostservices." group:"Host Services Configuration"`
 	OtelConfig         OtelConfig         `embed:"" prefix:"otel." group:"OpenTelemetry Configuration"`
@@ -435,6 +437,8 @@ func (u Up) Run(ctx context.Context, globals *Globals, n *Node) error {
 		options.WithValidIssuers(u.ValidIssuers),
 		options.WithOCICacheRegistry(u.OCICache),
 		options.WithDevMode(globals.DevMode),
+		options.WithStartWorkloadMessage(u.StartMessage),
+		options.WithStopWorkloadMessage(u.StopMessage),
 		options.WithOTelOptions(options.OTelOptions{
 			MetricsEnabled:   u.OtelConfig.OtelMetrics,
 			MetricsPort:      u.OtelConfig.OtelMetricsPort,

diff --git a/cmd/nex/workload.go b/cmd/nex/workload.go
@@ -17,9 +17,8 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
-	"time"
-
 	"strings"
+	"time"
 
 	"github.com/jedib0t/go-pretty/v6/table"
 	"github.com/synadia-io/nex/api/nodecontrol"
@@ -239,8 +238,14 @@ func (r RunWorkload) Run(ctx context.Context, globals *Globals, w *Workload) err
 		if resp.Started {
 			if r.NodeId != "" {
 				fmt.Printf("Workload %s [%s] started on node %s\n", r.WorkloadName, resp.Id, r.NodeId)
+				if resp.Message != "" {
+					fmt.Println(resp.Message)
+				}
 			} else {
 				fmt.Printf("Workload %s [%s] started\n", r.WorkloadName, resp.Id)
+				if resp.Message != "" {
+					fmt.Println(resp.Message)
+				}
 			}
 		} else {
 			fmt.Printf("Workload %s failed to start\n", r.WorkloadName)
@@ -289,6 +294,9 @@ func (s StopWorkload) Run(ctx context.Context, globals *Globals, w *Workload) er
 
 	if resp.Stopped {
 		fmt.Printf("Workload %s stopped\n", s.WorkloadId)
+		if resp.Message != "" {
+			fmt.Println(resp.Message)
+		}
 	} else {
 		fmt.Printf("Workload %s failed to stop\n", s.WorkloadId)
 	}
@@ -779,7 +787,6 @@ func (b BundleWorkload) Run(ctx context.Context, globals *Globals) error {
 		_, err = io.Copy(tw, file)
 		return err
 	})
-
 	if err != nil {
 		return err
 	}

diff --git a/models/node_options.go b/models/node_options.go
@@ -26,6 +26,9 @@ type NodeOptions struct {
 	OCICacheRegistry      string
 	DevMode               bool
 
+	StartWorkloadMessage string
+	StopWorkloadMessage  string
+
 	Errs error
 }
 
@@ -136,6 +139,18 @@ func WithDevMode(b bool) NodeOption {
 	}
 }
 
+func WithStartWorkloadMessage(s string) NodeOption {
+	return func(n *NodeOptions) {
+		n.StartWorkloadMessage = s
+	}
+}
+
+func WithStopWorkloadMessage(s string) NodeOption {
+	return func(n *NodeOptions) {
+		n.StopWorkloadMessage = s
+	}
+}
+
 type OTelOptions struct {
 	MetricsEnabled   bool
 	MetricsExporter  string

diff --git a/node/internal/actors/control_api.go b/node/internal/actors/control_api.go
@@ -61,6 +61,8 @@ type NodeCallback interface {
 	EncryptPayload([]byte, string) ([]byte, string, error)
 	DecryptPayload([]byte) ([]byte, error)
 	EmitEvent(string, json.RawMessage) error
+	StartWorkloadMessage() string
+	StopWorkloadMessage() string
 }
 
 type StateCallback interface {
@@ -363,8 +365,9 @@ func (api *ControlAPI) handleDeploy(m *nats.Msg) {
 		models.RespondEnvelope(m, RunResponseType, 500, "", fmt.Sprintf("failed to unmarshal workload started response: %s", err))
 		return
 	}
-
-	models.RespondEnvelope(m, RunResponseType, 200, startResponseFromProto(&workloadStarted), "")
+	resp := startResponseFromProto(&workloadStarted)
+	resp.Message = api.nodeCallback.StartWorkloadMessage()
+	models.RespondEnvelope(m, RunResponseType, 200, resp, "")
 }
 
 func (api *ControlAPI) handleUndeploy(m *nats.Msg) {
@@ -419,7 +422,9 @@ findWorkload:
 		models.RespondEnvelope(m, StopResponseType, 500, "", fmt.Sprintf("failed to unmarshal workload started response: %s", err))
 		return
 	}
-	models.RespondEnvelope(m, StopResponseType, 200, stopResponseFromProto(&workloadStopped), "")
+	resp := stopResponseFromProto(&workloadStopped)
+	resp.Message = api.nodeCallback.StopWorkloadMessage()
+	models.RespondEnvelope(m, StopResponseType, 200, resp, "")
 }
 
 func (api *ControlAPI) handleInfo(m *nats.Msg) {

diff --git a/node/node.go b/node/node.go
@@ -94,8 +94,10 @@ func NewNexNode(serverKey nkeys.KeyPair, nc *nats.Conn, opts ...models.NodeOptio
 			HostServiceOptions: models.HostServiceOptions{
 				Services: make(map[string]models.ServiceConfig),
 			},
-			OCICacheRegistry: "",
-			DevMode:          false,
+			OCICacheRegistry:     "",
+			DevMode:              false,
+			StartWorkloadMessage: "",
+			StopWorkloadMessage:  "",
 		},
 	}
 
@@ -719,3 +721,11 @@ func (nn *nexNode) getState() (map[string]*actorproto.StartWorkload, error) {
 
 	return reqs, nil
 }
+
+func (nn *nexNode) StartWorkloadMessage() string {
+	return nn.options.StartWorkloadMessage
+}
+
+func (nn *nexNode) StopWorkloadMessage() string {
+	return nn.options.StopWorkloadMessage
+}
diff --git a/test/copy_workload_test.go b/test/copy_workload_test.go
@@ -64,7 +64,7 @@ func TestCopyWorkload(t *testing.T) {
 
 		be.Equal(t, "ok", string(msg.Data))
 
-		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started$`)
+		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started\n.*$`)
 		match := re.FindStringSubmatch(strings.TrimSpace(origStdOut.String()))
 		be.Equal(t, 2, len(match))
 		origWorkloadId := match[1]
@@ -134,7 +134,7 @@ func TestMultipleCopyWorkload(t *testing.T) {
 		origDeploy.Stdout = origStdOut
 		be.NilErr(t, origDeploy.Run())
 
-		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started$`)
+		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started\n.*$`)
 		match := re.FindStringSubmatch(strings.TrimSpace(origStdOut.String()))
 		be.Equal(t, 2, len(match))
 		origWorkloadId := match[1]

diff --git a/test/helpers.go b/test/helpers.go
@@ -162,6 +162,6 @@ func startNexNodeCmd(t testing.TB, workingDir, nodeSeed, xkeySeed, natsServer, n
 		xkeySeed = string(xSeed)
 	}
 
-	cmd := exec.Command(cli, "node", "up", "--logger.level", "debug", "--logger.short", "-s", natsServer, "--resource-directory", workingDir, "--node-name", name, "--nexus", nexus, "--node-seed", nodeSeed, "--node-xkey-seed", xkeySeed)
+	cmd := exec.Command(cli, "node", "up", "--logger.level", "debug", "--logger.short", "-s", natsServer, "--resource-directory", workingDir, "--node-name", name, "--nexus", nexus, "--node-seed", nodeSeed, "--node-xkey-seed", xkeySeed, "--start-message", "test workload started", "--stop-message", "test workload stopped")
 	return cmd
 }
diff --git a/test/workload_test.go b/test/workload_test.go
@@ -83,7 +83,7 @@ func TestDirectStartService(t *testing.T) {
 
 		be.Equal(t, 0, len(cmdstderr.Bytes()))
 
-		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started on node (?P<node>[A-Z0-9]+)$`)
+		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started on node (?P<node>[A-Z0-9]+)\n.*$`)
 		match := re.FindStringSubmatch(strings.TrimSpace(cmdstdout.String()))
 		be.Equal(t, 3, len(match))
 		be.Equal(t, pub, match[2])
@@ -143,7 +143,7 @@ func TestDirectStartFunction(t *testing.T) {
 
 		be.Equal(t, 0, len(cmdstderr.Bytes()))
 
-		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started$`)
+		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started\n.*$`)
 		match := re.FindStringSubmatch(strings.TrimSpace(cmdstdout.String()))
 		be.Equal(t, 2, len(match))
 		time.Sleep(500 * time.Millisecond)
@@ -244,7 +244,7 @@ func TestDirectStop(t *testing.T) {
 
 		be.Equal(t, 0, len(cmdstderr.Bytes()))
 
-		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started$`)
+		re := regexp.MustCompile(`^Workload tester \[(?P<workload>[A-Za-z0-9]+)\] started\n.*$`)
 		match := re.FindStringSubmatch(strings.TrimSpace(cmdstdout.String()))
 		be.Equal(t, 2, len(match))
 
@@ -259,6 +259,7 @@ func TestDirectStop(t *testing.T) {
 		be.NilErr(t, cmd.Run())
 
 		be.In(t, "Workload "+workloadID+" stopped", cmdstdout.String())
+		be.In(t, "test workload stopped", cmdstdout.String())
 		cancel()
 	}()
 
@@ -297,6 +298,7 @@ func TestDirectStopNoWorkload(t *testing.T) {
 		be.NilErr(t, cmd.Run())
 
 		be.In(t, "Workload not found", cmdstdout.String())
+		be.NotIn(t, "test workload stopped", cmdstdout.String())
 		cancel()
 	}()