Fix regression related to AgentWorkloadInfo refactor (#339)

control-api/run.go

@@ -116,6 +116,12 @@ func NewDeployRequest(opts ...RequestOption) (*DeployRequest, error) {
 	return req, nil
 }
 
+// Returns true if the run request supports trigger subjects
+func (request *DeployRequest) SupportsTriggerSubjects() bool {
+	return request.WorkloadType == NexWorkloadV8 ||
+		request.WorkloadType == NexWorkloadWasm
+}
+
 // This will validate a request's workload JWT and return the parsed claims
 func (request *DeployRequest) Validate() (*jwt.GenericClaims, error) {
 	claims, err := jwt.DecodeGeneric(*request.WorkloadJWT)

control-api/utils.go

@@ -2,6 +2,7 @@ package controlapi
 
 import (
 	"encoding/base64"
+	"encoding/hex"
 	"strings"
 )
 
@@ -17,5 +18,5 @@ func SanitizeNATSDigest(input string) string {
 		return after
 	}
 
-	return string(h)
+	return hex.EncodeToString(h)
 }

internal/agent-api/client.go

@@ -27,7 +27,7 @@ type ContactLostCallback func(string)
 
 const (
 	defaultAgentPingIntervalMillis = 5000
-	maxRetryAttempts               = 3
+	maxRetryAttempts               = 10
 
 	NexTriggerSubject = "x-nex-trigger-subject"
 	NexRuntimeNs      = "x-nex-runtime-ns"

internal/agent-api/types.go

@@ -54,6 +54,7 @@ type AgentWorkloadInfo struct {
 	Essential     *bool             `json:"essential,omitempty"`
 	Hash          string            `json:"hash,omitempty"`
 	ID            *string           `json:"id"`
+	Location      *url.URL          `json:"location"`
 	Namespace     *string           `json:"namespace,omitempty"`
 	RetriedAt     *time.Time        `json:"retried_at,omitempty"`
 	RetryCount    *uint             `json:"retry_count,omitempty"`
@@ -68,7 +69,10 @@ type AgentWorkloadInfo struct {
 	Stdout      io.Writer `json:"-"`
 	TmpFilename *string   `json:"-"`
 
-	Location *url.URL `json:"-"`
+	EncryptedEnvironment *string `json:"-"`
+	JsDomain             *string `json:"-"`
+	SenderPublicKey      *string `json:"-"`
+	WorkloadJWT          *string `json:"-"`
 
 	Errors []error `json:"errors,omitempty"`
 }
@@ -109,6 +113,10 @@ func (r *AgentWorkloadInfo) Validate() error {
 		err = errors.Join(err, errors.New("hash is required"))
 	}
 
+	if r.Location == nil {
+		err = errors.Join(err, errors.New("location is required"))
+	}
+
 	if r.TotalBytes == 0 { // FIXME--- this should probably be checked against *string
 		err = errors.Join(err, errors.New("total bytes is required"))
 	}

internal/node/controlapi.go

@@ -284,8 +284,7 @@ func (api *ApiListener) handleDeploy(ctx context.Context, span trace.Span, m *na
 		return
 	}
 
-	if len(request.TriggerSubjects) > 0 && (request.WorkloadType != controlapi.NexWorkloadV8 &&
-		request.WorkloadType != controlapi.NexWorkloadWasm) { // FIXME -- workload type comparison
+	if len(request.TriggerSubjects) > 0 && !request.SupportsTriggerSubjects() { // FIXME -- workload type comparison
 		span.SetStatus(codes.Error, "Unsupported workload type for trigger subjects")
 		api.log.Error("Workload type does not support trigger subject registration", slog.String("trigger_subjects", string(request.WorkloadType)))
 		respondFail(controlapi.RunResponseType, m, fmt.Sprintf("Unsupported workload type for trigger subject registration: %s", string(request.WorkloadType)))
@@ -393,6 +392,7 @@ func (api *ApiListener) handleDeploy(ctx context.Context, span trace.Span, m *na
 			slog.String("namespace", namespace),
 			slog.String("workload", *agentDeployRequest.WorkloadName),
 			slog.String("workload_id", workloadID),
+			slog.String("workload_location", agentDeployRequest.Location.String()),
 			slog.Uint64("workload_size", numBytes),
 			slog.String("workload_sha256", *workloadHash),
 			slog.String("type", string(request.WorkloadType)),

internal/node/node.go

@@ -445,11 +445,14 @@ func (n *Node) handleAutostarts() {
 			continue
 		}
 
-		agentDeployRequest := agentWorkloadInfoFromControlDeployRequest(request, autostart.Namespace, numBytes, *workloadHash)
-		agentDeployRequest.TotalBytes = int64(numBytes)
-		agentDeployRequest.Hash = *workloadHash
+		agentWorkloadInfo := agentWorkloadInfoFromControlDeployRequest(request, autostart.Namespace, numBytes, *workloadHash)
 
-		err = n.manager.DeployWorkload(agentClient, agentDeployRequest)
+		agentWorkloadInfo.TotalBytes = int64(numBytes)
+		agentWorkloadInfo.Hash = *workloadHash
+
+		agentWorkloadInfo.Environment = autostart.Environment // HACK!!! we need to fix autostart config to allow encrypted environment...
+
+		err = n.manager.DeployWorkload(agentClient, agentWorkloadInfo)
 		if err != nil {
 			n.log.Error("Failed to deploy autostart workload",
 				slog.Any("error", err),
@@ -695,21 +698,24 @@ func (n *Node) shuttingDown() bool {
 
 func agentWorkloadInfoFromControlDeployRequest(request *controlapi.DeployRequest, namespace string, numBytes uint64, hash string) *agentapi.AgentWorkloadInfo {
 	return &agentapi.AgentWorkloadInfo{
-		Argv:               request.Argv,
-		DecodedClaims:      request.DecodedClaims,
-		Description:        request.Description,
-		Environment:        request.WorkloadEnvironment,
-		Essential:          request.Essential,
-		Hash:               hash,
-		HostServicesConfig: request.HostServicesConfig,
-		ID:                 request.ID,
-		Location:           request.Location,
-		Namespace:          &namespace,
-		RetriedAt:          request.RetriedAt,
-		RetryCount:         request.RetryCount,
-		TotalBytes:         int64(numBytes),
-		TriggerSubjects:    request.TriggerSubjects,
-		WorkloadName:       request.WorkloadName,
-		WorkloadType:       request.WorkloadType,
+		Argv:                 request.Argv,
+		DecodedClaims:        request.DecodedClaims,
+		Description:          request.Description,
+		EncryptedEnvironment: request.Environment,
+		Environment:          request.WorkloadEnvironment, // HACK!!! we need to fix autostart config to allow encrypted environment...
+		Essential:            request.Essential,
+		Hash:                 hash,
+		HostServicesConfig:   request.HostServicesConfig,
+		ID:                   request.ID,
+		Location:             request.Location,
+		Namespace:            &namespace,
+		RetriedAt:            request.RetriedAt,
+		RetryCount:           request.RetryCount,
+		SenderPublicKey:      request.SenderPublicKey,
+		TotalBytes:           int64(numBytes),
+		TriggerSubjects:      request.TriggerSubjects,
+		WorkloadJWT:          request.WorkloadJWT,
+		WorkloadName:         request.WorkloadName,
+		WorkloadType:         request.WorkloadType,
 	}
 }

internal/node/workload_mgr_events.go

@@ -94,17 +94,22 @@ func (w *WorkloadManager) agentEvent(agentId string, evt cloudevents.Event) {
 			digest := controlapi.SanitizeNATSDigest(info.Digest)
 
 			req, _ := json.Marshal(&controlapi.DeployRequest{
-				Argv:            agentWorkloadInfo.Argv,
-				Description:     agentWorkloadInfo.Description,
-				Hash:            &digest,
-				Essential:       agentWorkloadInfo.Essential,
-				ID:              &id,
-				Location:        agentWorkloadInfo.Location,
-				RetriedAt:       agentWorkloadInfo.RetriedAt,
-				RetryCount:      agentWorkloadInfo.RetryCount,
-				TriggerSubjects: agentWorkloadInfo.TriggerSubjects,
-				WorkloadName:    agentWorkloadInfo.WorkloadName,
-				WorkloadType:    agentWorkloadInfo.WorkloadType,
+				Argv:               agentWorkloadInfo.Argv,
+				Description:        agentWorkloadInfo.Description,
+				Hash:               &digest,
+				Environment:        agentWorkloadInfo.EncryptedEnvironment,
+				Essential:          agentWorkloadInfo.Essential,
+				HostServicesConfig: agentWorkloadInfo.HostServicesConfig,
+				ID:                 &id,
+				JsDomain:           agentWorkloadInfo.JsDomain,
+				Location:           agentWorkloadInfo.Location,
+				RetriedAt:          agentWorkloadInfo.RetriedAt,
+				RetryCount:         agentWorkloadInfo.RetryCount,
+				SenderPublicKey:    agentWorkloadInfo.SenderPublicKey,
+				TriggerSubjects:    agentWorkloadInfo.TriggerSubjects,
+				WorkloadJWT:        agentWorkloadInfo.WorkloadJWT,
+				WorkloadName:       agentWorkloadInfo.WorkloadName,
+				WorkloadType:       agentWorkloadInfo.WorkloadType,
 			})
 
 			nodeID := w.publicKey