Move log link filtering (by phase) from propeller to admin

Signed-off-by: Fabio Graetz <[email protected]>
flyteorg · Mar 18, 2024 · f41d673 · f41d673
1 parent 66c6b29
commit f41d673
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 35 deletions.
diff --git a/flyteadmin/pkg/repositories/transformers/task_execution.go b/flyteadmin/pkg/repositories/transformers/task_execution.go
@@ -372,6 +372,22 @@ func mergeMetadata(existing, latest *event.TaskExecutionMetadata) *event.TaskExe
 	return existing
 }
 
+func filterLogsByPhase(logs []*core.TaskLog, phase core.TaskExecution_Phase) []*core.TaskLog {
+	filteredLogs := make([]*core.TaskLog, 0, len(logs))
+
+	for _, l := range logs {
+		if common.IsTaskExecutionTerminal(phase) && l.HideOnceFinished {
+			continue
+		}
+		if phase == core.TaskExecution_QUEUED && !l.ShowWhilePending {
+			continue
+		}
+		filteredLogs = append(filteredLogs, l)
+
+	}
+	return filteredLogs
+}
+
 func UpdateTaskExecutionModel(ctx context.Context, request *admin.TaskExecutionEventRequest, taskExecutionModel *models.TaskExecution,
 	inlineEventDataPolicy interfaces.InlineEventDataPolicy, storageClient *storage.DataStore) error {
 	err := handleTaskExecutionInputs(ctx, taskExecutionModel, request, storageClient)
@@ -393,7 +409,11 @@ func UpdateTaskExecutionModel(ctx context.Context, request *admin.TaskExecutionE
 		reportedAt = request.Event.OccurredAt
 	}
 	taskExecutionClosure.UpdatedAt = reportedAt
-	taskExecutionClosure.Logs = mergeLogs(taskExecutionClosure.Logs, request.Event.Logs)
+
+	mergedLogs := mergeLogs(taskExecutionClosure.Logs, request.Event.Logs)
+	filteredLogs := filterLogsByPhase(mergedLogs, request.Event.Phase)
+	taskExecutionClosure.Logs = filteredLogs
+
 	if len(request.Event.Reasons) > 0 {
 		for _, reason := range request.Event.Reasons {
 			taskExecutionClosure.Reasons = append(

diff --git a/flyteplugins/go/tasks/pluginmachinery/core/phase.go b/flyteplugins/go/tasks/pluginmachinery/core/phase.go
@@ -227,16 +227,6 @@ func PhaseInfoQueued(t time.Time, version uint32, reason string) PhaseInfo {
 }
 
 func PhaseInfoQueuedWithTaskInfo(t time.Time, version uint32, reason string, info *TaskInfo) PhaseInfo {
-	if info != nil && info.Logs != nil {
-		logs := info.Logs
-		// Delete the logs for which ShowWhilePending is not true
-		info.Logs = make([]*core.TaskLog, 0, len(logs))
-		for _, l := range logs {
-			if l.ShowWhilePending {
-				info.Logs = append(info.Logs, l)
-			}
-		}
-	}
 	pi := phaseInfo(PhaseQueued, version, nil, info, false)
 	pi.reason = reason
 	return pi
@@ -249,7 +239,6 @@ func PhaseInfoInitializing(t time.Time, version uint32, reason string, info *Tas
 }
 
 func phaseInfoFailed(p Phase, err *core.ExecutionError, info *TaskInfo, cleanupOnFailure bool) PhaseInfo {
-	HideLogsOnceFinished(info)
 	if err == nil {
 		err = &core.ExecutionError{
 			Code:    "Unknown",
@@ -267,30 +256,7 @@ func PhaseInfoRunning(version uint32, info *TaskInfo) PhaseInfo {
 	return phaseInfo(PhaseRunning, version, nil, info, false)
 }
 
-func HideLogsOnceFinished(info *TaskInfo) {
-	if info != nil && info.Logs != nil {
-		logs := info.Logs
-		// Delete the logs for which hideOnceFinished is true
-		info.Logs = make([]*core.TaskLog, 0, len(logs))
-		for _, l := range logs {
-			if !l.HideOnceFinished {
-				info.Logs = append(info.Logs, l)
-			}
-		}
-	}
-}
-
 func PhaseInfoSuccess(info *TaskInfo) PhaseInfo {
-	if info != nil && info.Logs != nil {
-		logs := info.Logs
-		// Delete the logs for which hideOnceFinished is true
-		info.Logs = make([]*core.TaskLog, 0, len(logs))
-		for _, l := range logs {
-			if !l.HideOnceFinished {
-				info.Logs = append(info.Logs, l)
-			}
-		}
-	}
 	return phaseInfo(PhaseSuccess, DefaultPhaseVersion, nil, info, false)
 }