Track EventVersion in TaskExecutionClosure (flyteorg#395)

* updated flyteidl Signed-off-by: Daniel Rammer <[email protected]> * setting EventVersion on TaskExecutionClosure Signed-off-by: Daniel Rammer <[email protected]> * updated flyteidl Signed-off-by: Daniel Rammer <[email protected]>
eapolinario · Apr 8, 2022 · 9b1e129 · 9b1e129
1 parent 2f49a5a
commit 9b1e129
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 11 deletions.
diff --git a/flyteadmin/go.mod b/flyteadmin/go.mod
@@ -13,7 +13,7 @@ require (
 	github.com/cloudevents/sdk-go/v2 v2.8.0
 	github.com/coreos/go-oidc v2.2.1+incompatible
 	github.com/evanphx/json-patch v4.9.0+incompatible
-	github.com/flyteorg/flyteidl v0.24.15
+	github.com/flyteorg/flyteidl v0.24.17
 	github.com/flyteorg/flyteplugins v0.10.16
 	github.com/flyteorg/flytepropeller v0.16.36
 	github.com/flyteorg/flytestdlib v0.4.21

diff --git a/flyteadmin/go.sum b/flyteadmin/go.sum
@@ -369,8 +369,8 @@ github.com/felixge/httpsnoop v1.0.1 h1:lvB5Jl89CsZtGIWuTcDM1E/vkVs49/Ml7JJe07l8S
 github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/flyteorg/flyteidl v0.23.0/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
 github.com/flyteorg/flyteidl v0.24.0/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
-github.com/flyteorg/flyteidl v0.24.15 h1:Iqbwx3w1a4Dh6byRZrZMlHsKPKoOZbBiS9vR0iXzacY=
-github.com/flyteorg/flyteidl v0.24.15/go.mod h1:vHSugApgS3hRITIafzQDU8DZD/W8wFRfFcgaFU35Dww=
+github.com/flyteorg/flyteidl v0.24.17 h1:Xx70bJbuQGyvS8uAyU4AN74rot6KnzJ9r/L9gcCdEsU=
+github.com/flyteorg/flyteidl v0.24.17/go.mod h1:vHSugApgS3hRITIafzQDU8DZD/W8wFRfFcgaFU35Dww=
 github.com/flyteorg/flyteplugins v0.10.16 h1:rwNI2MACPbcST2O6CEUsNW6bccz7ZLni0GiY3orevfw=
 github.com/flyteorg/flyteplugins v0.10.16/go.mod h1:YBWV8QnFakDJfLyua8pYddiWqszAqseBKIJPNMERlos=
 github.com/flyteorg/flytepropeller v0.16.36 h1:5uE8JsutrPVyLVrRJ8BgvhZUOmTBFkEkn5xmIOo21nU=

diff --git a/flyteadmin/pkg/repositories/transformers/task_execution.go b/flyteadmin/pkg/repositories/transformers/task_execution.go
@@ -137,14 +137,15 @@ func CreateTaskExecutionModel(ctx context.Context, input CreateTaskExecutionMode
 	}
 
 	closure := &admin.TaskExecutionClosure{
-		Phase:      input.Request.Event.Phase,
-		UpdatedAt:  input.Request.Event.OccurredAt,
-		CreatedAt:  input.Request.Event.OccurredAt,
-		Logs:       input.Request.Event.Logs,
-		CustomInfo: input.Request.Event.CustomInfo,
-		Reason:     input.Request.Event.Reason,
-		TaskType:   input.Request.Event.TaskType,
-		Metadata:   metadata,
+		Phase:        input.Request.Event.Phase,
+		UpdatedAt:    input.Request.Event.OccurredAt,
+		CreatedAt:    input.Request.Event.OccurredAt,
+		Logs:         input.Request.Event.Logs,
+		CustomInfo:   input.Request.Event.CustomInfo,
+		Reason:       input.Request.Event.Reason,
+		TaskType:     input.Request.Event.TaskType,
+		Metadata:     metadata,
+		EventVersion: input.Request.Event.EventVersion,
 	}
 
 	eventPhase := input.Request.Event.Phase
@@ -374,6 +375,9 @@ func UpdateTaskExecutionModel(ctx context.Context, request *admin.TaskExecutionE
 		return errors.NewFlyteAdminErrorf(codes.Internal, "failed to merge task event custom_info with error: %v", err)
 	}
 	taskExecutionClosure.Metadata = mergeMetadata(taskExecutionClosure.Metadata, request.Event.Metadata)
+	if request.Event.EventVersion > taskExecutionClosure.EventVersion {
+		taskExecutionClosure.EventVersion = request.Event.EventVersion
+	}
 	marshaledClosure, err := proto.Marshal(&taskExecutionClosure)
 	if err != nil {
 		return errors.NewFlyteAdminErrorf(