[BUG] array node eventing bump version (flyteorg#5680)

* [BUG] add retries to handle array node eventing race condition (flyteorg#421)

If there is an error updating a [FlyteWorkflow CRD](https://github.com/unionai/flyte/blob/6a7207c5345604a28a9d4e3699becff767f520f5/flytepropeller/pkg/controller/handler.go#L378), then the propeller streak ends without the CRD getting updated and the in-memory copy of the FlyteWorkflow is not utilized on the next loop.

[TaskPhaseVersion](https://github.com/flyteorg/flyte/blob/37b4e13ac4a3594ac63b7a35058f4b2220e51282/flytepropeller/pkg/apis/flyteworkflow/v1alpha1/node_status.go#L239) is stored in the FlyteWorkflow. This is incremented when there is an update to node/subnode state to ensure that events are unique. If the events stay in the same state and have the same TaskPhaseVersion, then they [get short-circuited and don't get emitted to admin](https://github.com/flyteorg/flyte/blob/37b4e13ac4a3594ac63b7a35058f4b2220e51282/flytepropeller/events/admin_eventsink.go#L59) or will get returned as an [AlreadyExists error](https://github.com/flyteorg/flyte/blob/37b4e13ac4a3594ac63b7a35058f4b2220e51282/flyteadmin/pkg/manager/impl/task_execution_manager.go#L172) and get [handled in propeller to not bubble up in an error](https://github.com/flyteorg/flyte/blob/37b4e13ac4a3594ac63b7a35058f4b2220e51282/flytepropeller/pkg/controller/nodes/node_exec_context.go#L38).

We can run into issues with ArrayNode eventing when:
- array node handler increments task phase version from "0" to "1"
- admin event sink emits event with version "1"
- the propeller controller is not able to update the FlyteWorkflow CRD, so the ArrayNodeStatus indicates taskPhaseVersion is still 0
- next loop, array node handler increments task phase version from "0" to "1"
- admin event sink prevents the event from getting emitted as an event with the same ID has already been received. No error is bubbled up.

This means we lose subnode state until there is an event that contains an update to that subnode. If the lost state is the subnode reaching a terminal state, then the subnode state (from admin/UI) is "stuck" in a non-terminal state.

I confirmed this to be an issue in the load-test-cluster. Whenever, there was an [error syncing the FlyteWorkflow](https://github.com/flyteorg/flyte/blob/37b4e13ac4a3594ac63b7a35058f4b2220e51282/flytepropeller/pkg/controller/workers.go#L91), the next round of eventing in ArrayNode would fail unless the ArrayNode phase changed.

- added unit test
- tested locally in sandbox
- test in dogfood - https://buildkite.com/unionai/managed-cluster-staging-sync/builds/4398#01914a1a-f6d6-42a5-b41b-7b6807f27370

- should be fine to rollout to prod

Should this change be upstreamed to OSS (flyteorg/flyte)? If not, please uncheck this box, which is used for auditing. Note, it is the responsibility of each developer to actually upstream their changes. See [this guide](https://unionai.atlassian.net/wiki/spaces/ENG/pages/447610883/Flyte+-+Union+Cloud+Development+Runbook/#When-are-versions-updated%3F).
- [x] To be upstreamed to OSS

fixes: https://linear.app/unionai/issue/COR-1534/bug-arraynode-shows-non-complete-jobs-in-ui-when-the-job-is-actually

* [x] Added tests
* [x] Ran a deploy dry run and shared the terraform plan
* [ ] Added logging and metrics
* [ ] Updated [dashboards](https://unionai.grafana.net/dashboards) and [alerts](https://unionai.grafana.net/alerting/list)
* [ ] Updated documentation

Signed-off-by: Paul Dittamo <[email protected]>

* handle already exists error on array node abort (flyteorg#427)

* handle already exists error on array node abort

Signed-off-by: Paul Dittamo <[email protected]>

* update comment

Signed-off-by: Paul Dittamo <[email protected]>

---------

Signed-off-by: Paul Dittamo <[email protected]>

* [BUG] set cause for already exists EventError (flyteorg#432)

* set cause for already exists EventError

Signed-off-by: Paul Dittamo <[email protected]>

* add nil check event error

Signed-off-by: Paul Dittamo <[email protected]>

* lint

Signed-off-by: Paul Dittamo <[email protected]>

---------

Signed-off-by: Paul Dittamo <[email protected]>

---------

Signed-off-by: Paul Dittamo <[email protected]>
Signed-off-by: Bugra Gedik <[email protected]>

flytepropeller/events/admin_eventsink.go

@@ -57,7 +57,11 @@ func (s *adminEventSink) Sink(ctx context.Context, message proto.Message) error
 
 	if s.filter.Contains(ctx, id) {
 		logger.Debugf(ctx, "event '%s' has already been sent", string(id))
-		return nil
+		return &errors.EventError{
+			Code:    errors.AlreadyExists,
+			Cause:   fmt.Errorf("event has already been sent"),
+			Message: "Event Already Exists",
+		}
 	}
 
 	// Validate submission with rate limiter and send admin event

flytepropeller/events/admin_eventsink_test.go

@@ -184,13 +184,16 @@ func TestAdminFilterContains(t *testing.T) {
 	filter.OnContainsMatch(mock.Anything, mock.Anything).Return(true)
 
 	wfErr := adminEventSink.Sink(ctx, wfEvent)
-	assert.NoError(t, wfErr)
+	assert.Error(t, wfErr)
+	assert.True(t, errors.IsAlreadyExists(wfErr))
 
 	nodeErr := adminEventSink.Sink(ctx, nodeEvent)
-	assert.NoError(t, nodeErr)
+	assert.Error(t, nodeErr)
+	assert.True(t, errors.IsAlreadyExists(nodeErr))
 
 	taskErr := adminEventSink.Sink(ctx, taskEvent)
-	assert.NoError(t, taskErr)
+	assert.Error(t, taskErr)
+	assert.True(t, errors.IsAlreadyExists(taskErr))
 }
 
 func TestIDFromMessage(t *testing.T) {

flytepropeller/events/errors/errors.go

@@ -33,7 +33,11 @@ type EventError struct {
 }
 
 func (r EventError) Error() string {
-	return fmt.Sprintf("%s: %s, caused by [%s]", r.Code, r.Message, r.Cause.Error())
+	var cause string
+	if r.Cause != nil {
+		cause = r.Cause.Error()
+	}
+	return fmt.Sprintf("%s: %s, caused by [%s]", r.Code, r.Message, cause)
 }
 
 func (r *EventError) Is(target error) bool {

flytepropeller/pkg/controller/config/config.go

@@ -259,6 +259,7 @@ const (
 type EventConfig struct {
 	RawOutputPolicy           RawOutputPolicy `json:"raw-output-policy" pflag:",How output data should be passed along in execution events."`
 	FallbackToOutputReference bool            `json:"fallback-to-output-reference" pflag:",Whether output data should be sent by reference when it is too large to be sent inline in execution events."`
+	ErrorOnAlreadyExists      bool            `json:"error-on-already-exists" pflag:",Whether to return an error when an event already exists."`
 }
 
 // ParallelismBehavior defines how ArrayNode should handle subNode parallelism by default

flytepropeller/pkg/controller/nodes/array/handler.go

@@ -11,6 +11,7 @@ import (
 	"github.com/flyteorg/flyte/flyteplugins/go/tasks/pluginmachinery/ioutils"
 	"github.com/flyteorg/flyte/flyteplugins/go/tasks/plugins/array/errorcollector"
 	"github.com/flyteorg/flyte/flytepropeller/events"
+	eventsErr "github.com/flyteorg/flyte/flytepropeller/events/errors"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/compiler/validators"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/config"
@@ -21,6 +22,7 @@ import (
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/interfaces"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/nodes/task/k8s"
 	"github.com/flyteorg/flyte/flytestdlib/bitarray"
+	stdConfig "github.com/flyteorg/flyte/flytestdlib/config"
 	"github.com/flyteorg/flyte/flytestdlib/logger"
 	"github.com/flyteorg/flyte/flytestdlib/promutils"
 	"github.com/flyteorg/flyte/flytestdlib/storage"
@@ -112,6 +114,10 @@ func (a *arrayNodeHandler) Abort(ctx context.Context, nCtx interfaces.NodeExecut
 
 	// update state for subNodes
 	if err := eventRecorder.finalize(ctx, nCtx, idlcore.TaskExecution_ABORTED, 0, a.eventConfig); err != nil {
+		// a task event with abort phase is already emitted when handling ArrayNodePhaseFailing
+		if eventsErr.IsAlreadyExists(err) {
+			return nil
+		}
 		logger.Errorf(ctx, "ArrayNode event recording failed: [%s]", err.Error())
 		return err
 	}
@@ -579,12 +585,35 @@ func (a *arrayNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeExecu
 
 		// increment taskPhaseVersion if we detect any changes in subNode state.
 		if incrementTaskPhaseVersion {
-			arrayNodeState.TaskPhaseVersion = arrayNodeState.TaskPhaseVersion + 1
+			arrayNodeState.TaskPhaseVersion++
 		}
 
-		if err := eventRecorder.finalize(ctx, nCtx, taskPhase, arrayNodeState.TaskPhaseVersion, a.eventConfig); err != nil {
-			logger.Errorf(ctx, "ArrayNode event recording failed: [%s]", err.Error())
-			return handler.UnknownTransition, err
+		const maxRetries = 3
+		retries := 0
+		for retries <= maxRetries {
+			err := eventRecorder.finalize(ctx, nCtx, taskPhase, arrayNodeState.TaskPhaseVersion, a.eventConfig)
+
+			if err == nil {
+				break
+			}
+
+			// Handle potential race condition if FlyteWorkflow CRD fails to get synced
+			if eventsErr.IsAlreadyExists(err) {
+				if !incrementTaskPhaseVersion {
+					break
+				}
+				logger.Warnf(ctx, "Event version already exists, bumping version and retrying (%d/%d): [%s]", retries+1, maxRetries, err.Error())
+				arrayNodeState.TaskPhaseVersion++
+			} else {
+				logger.Errorf(ctx, "ArrayNode event recording failed: [%s]", err.Error())
+				return handler.UnknownTransition, err
+			}
+
+			retries++
+			if retries > maxRetries {
+				logger.Errorf(ctx, "ArrayNode event recording failed after %d retries: [%s]", maxRetries, err.Error())
+				return handler.UnknownTransition, err
+			}
 		}
 
 		// if the ArrayNode phase has changed we need to reset the taskPhaseVersion to 0
@@ -632,9 +661,21 @@ func New(nodeExecutor interfaces.Node, eventConfig *config.EventConfig, scope pr
 		return nil, err
 	}
 
+	eventConfigCopy, err := stdConfig.DeepCopyConfig(eventConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	deepCopiedEventConfig, ok := eventConfigCopy.(*config.EventConfig)
+	if !ok {
+		return nil, fmt.Errorf("deep copy error: expected *config.EventConfig, but got %T", eventConfigCopy)
+	}
+
+	deepCopiedEventConfig.ErrorOnAlreadyExists = true
+
 	arrayScope := scope.NewSubScope("array")
 	return &arrayNodeHandler{
-		eventConfig:                 eventConfig,
+		eventConfig:                 deepCopiedEventConfig,
 		gatherOutputsRequestChannel: make(chan *gatherOutputsRequest),
 		metrics:                     newMetrics(arrayScope),
 		nodeExecutionRequestChannel: make(chan *nodeExecutionRequest),