Merge remote-tracking branch 'origin/master' into artifacts-mono

Signed-off-by: Yee Hing Tong <[email protected]>
flyteorg · Sep 29, 2023 · 89ca9f9 · 89ca9f9
2 parents 832234b + b35cc95
commit 89ca9f9
Show file tree

Hide file tree

Showing 36 changed files with 3,517 additions and 241 deletions.
diff --git a/flyteadmin/pkg/manager/impl/task_execution_manager.go b/flyteadmin/pkg/manager/impl/task_execution_manager.go
@@ -189,7 +189,7 @@ func (m *TaskExecutionManager) CreateTaskExecutionEvent(ctx context.Context, req
 		return nil, err
 	}
 
-	if request.Event.Phase == core.TaskExecution_RUNNING && request.Event.PhaseVersion == 0 {
+	if request.Event.Phase == core.TaskExecution_RUNNING && request.Event.PhaseVersion == 0 { // TODO: need to be careful about missing inc/decs
 		m.metrics.ActiveTaskExecutions.Inc()
 	} else if common.IsTaskExecutionTerminal(request.Event.Phase) && request.Event.PhaseVersion == 0 {
 		m.metrics.ActiveTaskExecutions.Dec()

diff --git a/flyteadmin/pkg/manager/impl/testutils/mock_requests.go b/flyteadmin/pkg/manager/impl/testutils/mock_requests.go
@@ -300,3 +300,7 @@ func GetWorkflowRequestInterfaceBytes() []byte {
 	bytes, _ := proto.Marshal(GetWorkflowRequest().Spec.Template.Interface)
 	return bytes
 }
+
+func GetWorkflowRequestInterface() *core.TypedInterface {
+	return GetWorkflowRequest().Spec.Template.Interface
+}
diff --git a/flyteadmin/pkg/manager/impl/workflow_manager_test.go b/flyteadmin/pkg/manager/impl/workflow_manager_test.go
@@ -454,6 +454,13 @@ func TestListWorkflows(t *testing.T) {
 		assert.Equal(t, fmt.Sprintf("version %v", idx), workflow.Id.Version)
 		assert.True(t, proto.Equal(&admin.WorkflowClosure{
 			CreatedAt: testutils.MockCreatedAtProto,
+			CompiledWorkflow: &core.CompiledWorkflowClosure{
+				Primary: &core.CompiledWorkflow{
+					Template: &core.WorkflowTemplate{
+						Interface: &core.TypedInterface{},
+					},
+				},
+			},
 		}, workflow.Closure))
 	}
 	assert.Empty(t, workflowList.Token)

diff --git a/flyteadmin/pkg/repositories/transformers/task_execution.go b/flyteadmin/pkg/repositories/transformers/task_execution.go
@@ -153,19 +153,27 @@ func CreateTaskExecutionModel(ctx context.Context, input CreateTaskExecutionMode
 		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,
 	}
 
-	if len(input.Request.Event.Reason) > 0 {
+	if len(input.Request.Event.Reasons) > 0 {
+		for _, reason := range input.Request.Event.Reasons {
+			closure.Reasons = append(closure.Reasons, &admin.Reason{
+				OccurredAt: reason.OccurredAt,
+				Message:    reason.Reason,
+			})
+		}
+		closure.Reason = input.Request.Event.Reasons[len(input.Request.Event.Reasons)-1].Reason
+	} else if len(input.Request.Event.Reason) > 0 {
 		closure.Reasons = []*admin.Reason{
-			&admin.Reason{
+			{
 				OccurredAt: input.Request.Event.OccurredAt,
 				Message:    input.Request.Event.Reason,
 			},
 		}
+		closure.Reason = input.Request.Event.Reason
 	}
 
 	eventPhase := input.Request.Event.Phase
@@ -386,7 +394,17 @@ func UpdateTaskExecutionModel(ctx context.Context, request *admin.TaskExecutionE
 	}
 	taskExecutionClosure.UpdatedAt = reportedAt
 	taskExecutionClosure.Logs = mergeLogs(taskExecutionClosure.Logs, request.Event.Logs)
-	if len(request.Event.Reason) > 0 {
+	if len(request.Event.Reasons) > 0 {
+		for _, reason := range request.Event.Reasons {
+			taskExecutionClosure.Reasons = append(
+				taskExecutionClosure.Reasons,
+				&admin.Reason{
+					OccurredAt: reason.OccurredAt,
+					Message:    reason.Reason,
+				})
+		}
+		taskExecutionClosure.Reason = request.Event.Reasons[len(request.Event.Reasons)-1].Reason
+	} else if len(request.Event.Reason) > 0 {
 		if taskExecutionClosure.Reason != request.Event.Reason {
 			// by tracking a time-series of reasons we increase the size of the TaskExecutionClosure in scenarios where
 			// a task reports a large number of unique reasons. if this size increase becomes problematic we this logic