Update metadata in ArrayNode TaskExecutionEvents (#4355)

* adding PluginIdentifier to TaskExecutionMetadata Signed-off-by: Daniel Rammer <[email protected]> * sending input and output metadata - removing validating check on flyteadmin temporarily Signed-off-by: Daniel Rammer <[email protected]> * readded task execution id verification Signed-off-by: Daniel Rammer <[email protected]> * removed dead code Signed-off-by: Daniel Rammer <[email protected]> --------- Signed-off-by: Daniel Rammer <[email protected]>
flyteorg · Nov 8, 2023 · dd91cba · dd91cba
1 parent 630724f
commit dd91cba
Showing 1 changed file with 28 additions and 0 deletions.
diff --git a/flytepropeller/pkg/controller/nodes/array/event_recorder.go b/flytepropeller/pkg/controller/nodes/array/event_recorder.go
@@ -120,11 +120,39 @@ func (e *externalResourcesEventRecorder) finalize(ctx context.Context, nCtx inte
 		OccurredAt:            occurredAt,
 		Metadata: &event.TaskExecutionMetadata{
 			ExternalResources: e.externalResources,
+			PluginIdentifier:  "container",
 		},
 		TaskType:     "k8s-array",
 		EventVersion: 1,
 	}
 
+	// only attach input values if taskPhase is QUEUED meaning this the first evaluation
+	if taskPhase == idlcore.TaskExecution_QUEUED {
+		if eventConfig.RawOutputPolicy == config.RawOutputPolicyInline {
+			// pass inputs by value
+			literalMap, err := nCtx.InputReader().Get(ctx)
+			if err != nil {
+				return err
+			}
+
+			taskExecutionEvent.InputValue = &event.TaskExecutionEvent_InputData{
+				InputData: literalMap,
+			}
+		} else {
+			// pass inputs by reference
+			taskExecutionEvent.InputValue = &event.TaskExecutionEvent_InputUri{
+				InputUri: nCtx.InputReader().GetInputPath().String(),
+			}
+		}
+	}
+
+	// only attach output uri if taskPhase is SUCCEEDED
+	if taskPhase == idlcore.TaskExecution_SUCCEEDED {
+		taskExecutionEvent.OutputResult = &event.TaskExecutionEvent_OutputUri{
+			OutputUri: v1alpha1.GetOutputsFile(nCtx.NodeStatus().GetOutputDir()).String(),
+		}
+	}
+
 	// record TaskExecutionEvent
 	return e.EventRecorder.RecordTaskEvent(ctx, taskExecutionEvent, eventConfig)
 }