fix conflicts

Signed-off-by: Yee Hing Tong <[email protected]>

flyteidl/protos/flyteidl/event/event.proto

@@ -115,20 +115,12 @@ message NodeExecutionEvent {
     // Indicates if this node is an ArrayNode.
     bool is_array = 22;
 
-<<<<<<< HEAD
-    // Holding this field here for now, this will be upstreamed soon.
-=======
->>>>>>> 5ec9fe3cc (Add fields to NodeExecutionEvent (#315))
     // So that Admin doesn't have to rebuild the node execution graph to find the target entity, propeller will fill this
     // in optionally - currently this is only filled in for subworkflows. This is the ID of the subworkflow corresponding
     // to this node execution. It is difficult to find because Admin only sees one node at a time. A subworkflow could be
     // nested multiple layers deep, and you'd need to access the correct workflow template to know the target subworkflow.
     core.Identifier target_entity = 23;
 
-<<<<<<< HEAD
-    // Holding this field here for now, this will be upstreamed soon.
-=======
->>>>>>> 5ec9fe3cc (Add fields to NodeExecutionEvent (#315))
     // Tasks and subworkflows (but not launch plans) that are run within a dynamic task are effectively independent of
     // the tasks that are registered in Admin's db. Confusingly, they are often identical, but sometimes they are not
     // even registered at all. Similar to the target_entity field, at the time Admin receives this event, it has no idea

flytepropeller/pkg/controller/nodes/common/utils.go

@@ -49,8 +49,10 @@ func GetTargetEntity(ctx context.Context, nCtx interfaces.NodeExecutionContext)
 	if nCtx.Node().GetWorkflowNode() != nil {
 		subRef := nCtx.Node().GetWorkflowNode().GetSubWorkflowRef()
 		if subRef != nil && len(*subRef) > 0 {
-			subWorkflow := nCtx.ExecutionContext().FindSubWorkflow(*subRef)
-			targetEntity = subWorkflow.GetIdentifier()
+			// todo: uncomment this if Support caching subworkflows and launchplans (v2) is upstreamed
+			// for now, we can leave it empty
+			//nCtx.ExecutionContext().FindSubWorkflow(*subRef)
+			//targetEntity = subWorkflow.GetIdentifier()
 		} else if nCtx.Node().GetWorkflowNode().GetLaunchPlanRefID() != nil {
 			lpRef := nCtx.Node().GetWorkflowNode().GetLaunchPlanRefID()
 			targetEntity = lpRef.Identifier

flytepropeller/pkg/controller/nodes/executor_test.go

@@ -1682,16 +1682,11 @@ func TestNodeExecutor_AbortHandler(t *testing.T) {
 		execContext := mocks4.ExecutionContext{}
 		execContext.EXPECT().IsInterruptible().Return(false)
 		r := v1alpha1.RawOutputDataConfig{}
-<<<<<<< HEAD
 		execContext.EXPECT().GetRawOutputDataConfig().Return(r)
 		execContext.EXPECT().GetExecutionID().Return(v1alpha1.WorkflowExecutionIdentifier{})
 		execContext.EXPECT().GetLabels().Return(nil)
 		execContext.EXPECT().GetEventVersion().Return(v1alpha1.EventVersion0)
-=======
-		execContext.OnGetRawOutputDataConfig().Return(r)
-		execContext.OnGetExecutionID().Return(v1alpha1.WorkflowExecutionIdentifier{})
-		execContext.OnGetLabels().Return(nil)
-		execContext.OnGetEventVersion().Return(v1alpha1.EventVersion0)
+
 		et := &mocks.ExecutableTask{}
 		et.OnCoreTask().Return(&core.TaskTemplate{
 			Id: &core.Identifier{
@@ -1702,13 +1697,12 @@ func TestNodeExecutor_AbortHandler(t *testing.T) {
 				Version:      "v",
 			},
 		})
-		execContext.OnGetTask("id").Return(et, nil)
+		execContext.EXPECT().GetTask("id").Return(et, nil)
 		parentInfo := &mocks4.ImmutableParentInfo{}
 		parentInfo.OnGetUniqueID().Return("someunique1")
 		parentInfo.OnCurrentAttempt().Return(uint32(1))
 		parentInfo.OnIsInDynamicChain().Return(false)
-		execContext.OnGetParentInfo().Return(parentInfo)
->>>>>>> 5ec9fe3cc (Add fields to NodeExecutionEvent (#315))
+		execContext.EXPECT().GetParentInfo().Return(parentInfo)
 
 		assert.NoError(t, nExec.AbortHandler(ctx, &execContext, &dag, nl, n, "aborting"))
 	})

flytepropeller/pkg/controller/nodes/node_exec_context_test.go

@@ -153,37 +153,6 @@ func Test_NodeContextDefault(t *testing.T) {
 	assert.Equal(t, w1.Tasks["taskID"].TaskTemplate.Id.Version, taskIdentifier.Version)
 }
 
-func TestGetTargetEntity_SubWorkflowNode(t *testing.T) {
-	id := &core.Identifier{
-		ResourceType: core.ResourceType_WORKFLOW,
-		Project:      "proj",
-		Domain:       "domain",
-		Name:         "sub-sub",
-		Version:      "v1",
-	}
-	exSubWf := &mocks.ExecutableSubWorkflow{}
-	exSubWf.OnGetIdentifier().Return(id)
-	ec := mocks2.ExecutionContext{}
-	ec.OnFindSubWorkflow("sub-workflow").Return(exSubWf)
-
-	subWfNode := &mocks.ExecutableWorkflowNode{}
-	subWfID := "sub-workflow"
-	subWfNode.OnGetSubWorkflowRef().Return(&subWfID)
-
-	n := &mocks.ExecutableNode{}
-	n.OnGetWorkflowNode().Return(subWfNode)
-
-	nCtx := &mocks3.NodeExecutionContext{}
-	nCtx.OnNode().Return(n)
-	nCtx.OnExecutionContext().Return(&ec)
-
-	fetchedID := common.GetTargetEntity(context.Background(), nCtx)
-	assert.Equal(t, id.Project, fetchedID.Project)
-	assert.Equal(t, id.Domain, fetchedID.Domain)
-	assert.Equal(t, id.Name, fetchedID.Name)
-	assert.Equal(t, id.Version, fetchedID.Version)
-}
-
 func TestGetTargetEntity_LaunchPlanNode(t *testing.T) {
 	id := &core.Identifier{
 		ResourceType: core.ResourceType_LAUNCH_PLAN,
@@ -210,38 +179,6 @@ func TestGetTargetEntity_LaunchPlanNode(t *testing.T) {
 	assert.Equal(t, id.Version, fetchedID.Version)
 }
 
-func TestGetTargetEntity_Task(t *testing.T) {
-	id := &core.Identifier{
-		ResourceType: core.ResourceType_TASK,
-		Project:      "proj",
-		Domain:       "domain",
-		Name:         "task",
-		Version:      "v3",
-	}
-
-	n := &mocks.ExecutableNode{}
-	n.OnGetWorkflowNode().Return(nil)
-	taskID := "task-id"
-	n.OnGetTaskID().Return(&taskID)
-
-	taskTemplate := &core.TaskTemplate{Id: id}
-
-	exTask := &mocks.ExecutableTask{}
-	exTask.OnCoreTask().Return(taskTemplate)
-	ec := mocks2.ExecutionContext{}
-	ec.OnGetTask(taskID).Return(exTask, nil)
-
-	nCtx := &mocks3.NodeExecutionContext{}
-	nCtx.OnNode().Return(n)
-	nCtx.OnExecutionContext().Return(&ec)
-
-	fetchedID := common.GetTargetEntity(context.Background(), nCtx)
-	assert.Equal(t, id.Project, fetchedID.Project)
-	assert.Equal(t, id.Domain, fetchedID.Domain)
-	assert.Equal(t, id.Name, fetchedID.Name)
-	assert.Equal(t, id.Version, fetchedID.Version)
-}
-
 func TestGetTargetEntity_EmptyTask(t *testing.T) {
 	n := &mocks.ExecutableNode{}
 	n.OnGetWorkflowNode().Return(nil)