Merge pull request #737 from ystia/backport40/bugfix/gh-733-action-no-stop

[Backport 4.0] Workflow with asynchronous action never stops after another step failure

Author: laurentganne

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## UNRELEASED
 
+### BUG FIXES
+
+* Workflow with asynchronous action never stops after another step failure  ([GH-733](https://github.com/ystia/yorc/issues/733))
+
 ## 4.0.6 (May 06, 2021)
 
 ### BUG FIXES

prov/scheduling/scheduler/consul_test.go

@@ -71,5 +71,8 @@ func TestRunConsulSchedulingPackageTests(t *testing.T) {
 		t.Run("testUnregisterAction", func(t *testing.T) {
 			testUnregisterAction(t, client)
 		})
+		t.Run("testUpdateActionData", func(t *testing.T) {
+			testUpdateActionData(t, client)
+		})
 	})
 }

prov/scheduling/scheduler/scheduler_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/hashicorp/consul/api"
 	"github.com/stretchr/testify/require"
+	"gotest.tools/v3/assert"
 
 	"github.com/ystia/yorc/v4/events"
 	"github.com/ystia/yorc/v4/helper/consulutil"
@@ -368,3 +369,30 @@ func testUnregisterAction(t *testing.T, client *api.Client) {
 	require.NotNil(t, kvp, "kvp is nil")
 	require.Equal(t, "true", string(kvp.Value), "unregisterFlag is not set to true")
 }
+
+func testUpdateActionData(t *testing.T, client *api.Client) {
+	t.Parallel()
+	deploymentID := "dep-" + t.Name()
+	ti := 1 * time.Second
+	actionType := "test-action"
+	action := &prov.Action{ActionType: actionType, Data: map[string]string{"key1": "val1", "key2": "val2", "key3": "val3"}}
+	id, err := scheduling.RegisterAction(client, deploymentID, ti, action)
+	assert.NilError(t, err, "Failed to register action")
+
+	err = scheduling.UpdateActionData(client, id, "key2", "newVal")
+	assert.NilError(t, err, "Failed to update action data")
+
+	testSched := scheduler{cc: client}
+	newAction, err := testSched.buildScheduledAction(id)
+	assert.NilError(t, err, "Failed to build action")
+
+	val := newAction.Data["key2"]
+	assert.Equal(t, val, "newVal", "Unexpected value for action key updated")
+
+	// Check the update of an unregistered action, should fail
+	err = testSched.unregisterAction(id)
+	assert.NilError(t, err, "Failed to unregister action")
+
+	err = scheduling.UpdateActionData(client, id, "key3", "newVal")
+	assert.ErrorContains(t, err, "unregistered")
+}

prov/scheduling/scheduling.go

@@ -22,7 +22,7 @@ import (
 
 	"github.com/hashicorp/consul/api"
 	"github.com/pkg/errors"
-	"github.com/satori/go.uuid"
+	uuid "github.com/satori/go.uuid"
 
 	"github.com/ystia/yorc/v4/helper/consulutil"
 	"github.com/ystia/yorc/v4/log"
@@ -104,7 +104,16 @@ func UnregisterAction(client *api.Client, id string) error {
 // UpdateActionData updates the value of a given data within an action
 func UpdateActionData(client *api.Client, id, key, value string) error {
 
-	//TODO check if action exists
-	scaKeyPath := path.Join(consulutil.SchedulingKVPrefix, "actions", id, "data", key)
+	// check if action still exists
+	actionIdPrefix := path.Join(consulutil.SchedulingKVPrefix, "actions", id)
+	kvp, _, err := client.KV().Get(path.Join(actionIdPrefix, "deploymentID"), nil)
+	if err != nil {
+		return err
+	}
+	if kvp == nil {
+		return errors.Errorf("Action with ID %s is unregistered", id)
+	}
+
+	scaKeyPath := path.Join(actionIdPrefix, "data", key)
 	return errors.Wrapf(consulutil.StoreConsulKeyAsString(scaKeyPath, value), "Failed to update data %q for action %q", key, id)
 }

tasks/workflow/worker.go

@@ -448,6 +448,7 @@ func (w *worker) runAction(ctx context.Context, t *taskExecution) error {
 		}
 	}
 	wasCancelled := new(bool)
+	taskFailure := new(bool)
 	if action.AsyncOperation.TaskID != "" {
 		ctx = operations.SetOperationLogFields(ctx, action.AsyncOperation.Operation)
 		ctx = events.AddLogOptionalFields(ctx, events.LogOptionalFields{
@@ -468,6 +469,7 @@ func (w *worker) runAction(ctx context.Context, t *taskExecution) error {
 			tasks.UpdateTaskStepWithStatus(action.AsyncOperation.TaskID, action.AsyncOperation.StepName, tasks.TaskStepStatusCANCELED)
 		})
 		tasks.MonitorTaskFailure(ctx, action.AsyncOperation.TaskID, func() {
+			*taskFailure = true
 			// Unregister this action asap to prevent new schedulings
 			scheduling.UnregisterAction(w.consulClient, action.ID)
 
@@ -496,6 +498,9 @@ func (w *worker) runAction(ctx context.Context, t *taskExecution) error {
 	if deregister || *wasCancelled {
 		scheduling.UnregisterAction(w.consulClient, action.ID)
 		w.endAction(ctx, t, action, *wasCancelled, err)
+	} else if *taskFailure {
+		err = errors.Errorf("Stopped on task failure")
+		w.endAction(ctx, t, action, *wasCancelled, err)
 	}
 	if err != nil {
 		return err