wip: ts only task updates

Author: joelrebel

pkg/api/v1/client/client_orchestrator.go

@@ -44,9 +44,17 @@ func (c *Client) ControllerCheckin(ctx context.Context, serverID, conditionID uu
 	return c.get(ctx, path)
 }
 
-func (c *Client) ConditionTaskPublish(ctx context.Context, conditionKind rctypes.Kind, serverID, conditionID uuid.UUID, task *rctypes.Task[any, any]) (*v1types.ServerResponse, error) {
+func (c *Client) ConditionTaskPublish(ctx context.Context, conditionKind rctypes.Kind, serverID, conditionID uuid.UUID, task *rctypes.Task[any, any], onlyUpdateTimestamp bool) (*v1types.ServerResponse, error) {
 	path := fmt.Sprintf("servers/%s/condition-task/%s/%s", serverID.String(), conditionKind, conditionID.String())
 
+	if onlyUpdateTimestamp {
+		path += "&ts_update=true"
+	} else {
+		if task == nil {
+			task = &rctypes.Task[any, any]{}
+		}
+	}
+
 	return c.post(ctx, path, task)
 }
 

pkg/api/v1/routes/handlers_orchestrator.go

@@ -22,6 +22,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	v1types "github.com/metal-toolbox/conditionorc/pkg/api/v1/types"
+	"github.com/metal-toolbox/rivets/condition"
 	rctypes "github.com/metal-toolbox/rivets/condition"
 	rkv "github.com/metal-toolbox/rivets/events/pkg/kv"
 	"github.com/metal-toolbox/rivets/events/registry"
@@ -165,21 +166,11 @@ func (r *Routes) conditionQueuePop(c *gin.Context) (int, *v1types.ServerResponse
 		}
 	}
 
-	// create Task entry
-	task := &rctypes.Task[any, any]{
-		StructVersion: rctypes.TaskVersion1,
-		ID:            cond.ID,
-		Kind:          conditionKind,
-		State:         rctypes.Pending,
-		Status:        st,
-		WorkerID:      controllerID.String(),
-		Asset:         &rctypes.Asset{ID: serverID},
-		TraceID:       trace.SpanFromContext(ctx).SpanContext().TraceID().String(),
-		SpanID:        trace.SpanFromContext(ctx).SpanContext().SpanID().String(),
-		Parameters:    cond.Parameters,
-		Data:          json.RawMessage(`{"empty": true}`), // placeholder value
-		Fault:         cond.Fault,
-	}
+	// setup task to be published
+	task := condition.NewTaskFromCondition(cond)
+	task.WorkerID = controllerID.String()
+	task.TraceID = trace.SpanFromContext(ctx).SpanContext().TraceID().String()
+	task.SpanID = trace.SpanFromContext(ctx).SpanContext().SpanID().String()
 
 	if err := r.kvPublishTask(
 		ctx,
@@ -188,6 +179,7 @@ func (r *Routes) conditionQueuePop(c *gin.Context) (int, *v1types.ServerResponse
 		conditionKind,
 		task,
 		true,
+		false,
 	); err != nil {
 		r.logger.WithField("condition.id", cond.ID).WithError(err).Info("task KV publish error")
 
@@ -409,7 +401,7 @@ func (r *Routes) kvPublishStatusValue(
 	controllerID registry.ControllerID,
 	conditionKind rctypes.Kind,
 	newSV *rctypes.StatusValue,
-	create bool,
+	create,
 	onlyTimestamp bool,
 ) error {
 	statusKV, err := status.GetConditionKV(conditionKind)
@@ -763,15 +755,22 @@ func (r *Routes) taskPublish(c *gin.Context) (int, *v1types.ServerResponse) {
 		attribute.KeyValue{Key: "serverId", Value: attribute.StringValue(c.Param("uuid"))},
 		attribute.KeyValue{Key: "conditionKind", Value: attribute.StringValue(c.Param("conditionKind"))},
 		attribute.KeyValue{Key: "conditionID", Value: attribute.StringValue(c.Param("conditionID"))},
+		attribute.KeyValue{Key: "timestampUpdate", Value: attribute.StringValue(c.Request.URL.Query().Get("ts_update"))},
 	)
 	defer span.End()
 
 	var task rctypes.Task[any, any]
-	if err := c.ShouldBindJSON(&task); err != nil {
-		r.logger.WithError(err).Warn("unmarshal Task payload")
+	var onlyTimestampUpdate bool
 
-		return http.StatusBadRequest, &v1types.ServerResponse{
-			Message: "invalid Task payload: " + err.Error(),
+	if c.Request.URL.Query().Get("ts_update") == "true" {
+		onlyTimestampUpdate = true
+	} else {
+		if err := c.ShouldBindJSON(&task); err != nil {
+			r.logger.WithError(err).Warn("unmarshal Task payload")
+
+			return http.StatusBadRequest, &v1types.ServerResponse{
+				Message: "invalid Task payload: " + err.Error(),
+			}
 		}
 	}
 
@@ -829,6 +828,7 @@ func (r *Routes) taskPublish(c *gin.Context) (int, *v1types.ServerResponse) {
 		conditionKind,
 		&task,
 		false,
+		onlyTimestampUpdate,
 	); err != nil {
 		return http.StatusInternalServerError, &v1types.ServerResponse{
 			Message: "Task publish error: " + err.Error(),
@@ -873,7 +873,15 @@ func (r *Routes) registerRegistryKVHandle() error {
 }
 
 // Publish implements the ConditionTaskRepository interface, it transparently publishes the given json.RawMessage to the KV
-func (r *Routes) kvPublishTask(ctx context.Context, serverID, conditionID string, conditionKind rctypes.Kind, task *rctypes.Task[any, any], create bool) error {
+func (r *Routes) kvPublishTask(
+	ctx context.Context,
+	serverID,
+	conditionID string,
+	conditionKind rctypes.Kind,
+	task *rctypes.Task[any, any],
+	create,
+	onlyTimestamp bool,
+) error {
 	_, span := otel.Tracer(pkgName).Start(
 		ctx,
 		"controller.Publish.KV.Task",
@@ -947,24 +955,20 @@ func (r *Routes) kvPublishTask(ctx context.Context, serverID, conditionID string
 		return failed(errors.Wrap(err, "Task object deserialize error"))
 	}
 
-	// verify conditionID matches before update
-	if currTask.ID.String() != conditionID && !rctypes.StateIsComplete(currTask.State) {
-		msg := fmt.Errorf(
-			"existing Task object %s in in-complete state, does not match conditionID: %s, must be purged before proceeding",
-			currTask.ID.String(),
-			conditionID,
-		)
-
-		return failed(msg)
+	if onlyTimestamp {
+		currTask.UpdatedAt = time.Now()
+	} else {
+		if err := currTask.Update(task); err != nil {
+			return failed(err)
+		}
 	}
 
-	task.UpdatedAt = time.Now()
-	taskJSON, err := task.Marshal()
+	currTaskJSON, err := currTask.Marshal()
 	if err != nil {
 		return failed(err)
 	}
 
-	if _, err := taskKV.Update(key, taskJSON, curr.Revision()); err != nil {
+	if _, err := taskKV.Update(key, currTaskJSON, curr.Revision()); err != nil {
 		return failed(err)
 	}