Update metadata in ArrayNode TaskExecutionEvents

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

@@ -120,11 +120,39 @@
 		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)
 }