fix: taskresults owned by pod rather than workflow. (#8284)

Signed-off-by: Alex Collins <alex_collins@intuit.com>

cmd/argoexec/commands/root.go

@@ -123,10 +123,10 @@ func initExecutor() *executor.WorkflowExecutor {
 		versioned.NewForConfigOrDie(config).ArgoprojV1alpha1().WorkflowTaskResults(namespace),
 		restClient,
 		podName,
+		types.UID(os.Getenv(common.EnvVarPodUID)),
 		os.Getenv(common.EnvVarWorkflowName),
 		os.Getenv(common.EnvVarNodeID),
 		namespace,
-		types.UID(os.Getenv(common.EnvVarWorkflowUID)),
 		cre,
 		*tmpl,
 		includeScriptOutput,

workflow/common/common.go

@@ -103,12 +103,12 @@ const (
 
 	// EnvVarPodName contains the name of the pod (currently unused)
 	EnvVarPodName = "ARGO_POD_NAME"
+	// EnvVarPodUID is the workflow's UID
+	EnvVarPodUID = "ARGO_POD_UID"
 	// EnvVarInstanceID is the instance ID
 	EnvVarInstanceID = "ARGO_INSTANCE_ID"
 	// EnvVarWorkflowName is the name of the workflow for which the an agent is responsible for
 	EnvVarWorkflowName = "ARGO_WORKFLOW_NAME"
-	// EnvVarWorkflowUID is the workflow's UID
-	EnvVarWorkflowUID = "ARGO_WORKFLOW_UID"
 	// EnvVarNodeID is the node ID of the node.
 	EnvVarNodeID = "ARGO_NODE_ID"
 	// EnvVarPluginAddresses is a list of plugin addresses

workflow/controller/workflowpod.go

@@ -572,6 +572,15 @@ func (woc *wfOperationCtx) createEnvVars() []apiv1.EnvVar {
 				},
 			},
 		},
+		{
+			Name: common.EnvVarPodUID,
+			ValueFrom: &apiv1.EnvVarSource{
+				FieldRef: &apiv1.ObjectFieldSelector{
+					APIVersion: "v1",
+					FieldPath:  "metadata.uid",
+				},
+			},
+		},
 		{
 			Name:  common.EnvVarContainerRuntimeExecutor,
 			Value: woc.getContainerRuntimeExecutor(),
@@ -588,10 +597,6 @@ func (woc *wfOperationCtx) createEnvVars() []apiv1.EnvVar {
 			Name:  common.EnvVarWorkflowName,
 			Value: woc.wf.Name,
 		},
-		{
-			Name:  common.EnvVarWorkflowUID,
-			Value: string(woc.wf.UID),
-		},
 	}
 	if v := woc.controller.Config.InstanceID; v != "" {
 		execEnvVars = append(execEnvVars,

workflow/executor/executor.go

@@ -55,8 +55,8 @@ const (
 // WorkflowExecutor is program which runs as the init/wait container
 type WorkflowExecutor struct {
 	PodName             string
+	podUID              types.UID
 	workflow            string
-	workflowUID         types.UID
 	nodeId              string
 	Template            wfv1.Template
 	IncludeScriptOutput bool
@@ -113,10 +113,11 @@ type ContainerRuntimeExecutor interface {
 // NewExecutor instantiates a new workflow executor
 func NewExecutor(
 	clientset kubernetes.Interface,
-	taskSetClient argoprojv1.WorkflowTaskResultInterface,
+	taskResultClient argoprojv1.WorkflowTaskResultInterface,
 	restClient rest.Interface,
-	podName, workflow, nodeId, namespace string,
-	workflowUID types.UID,
+	podName string,
+	podUID types.UID,
+	workflow, nodeId, namespace string,
 	cre ContainerRuntimeExecutor,
 	template wfv1.Template,
 	includeScriptOutput bool,
@@ -126,11 +127,11 @@ func NewExecutor(
 	log.WithFields(log.Fields{"Steps": executorretry.Steps, "Duration": executorretry.Duration, "Factor": executorretry.Factor, "Jitter": executorretry.Jitter}).Info("Using executor retry strategy")
 	return WorkflowExecutor{
 		PodName:                      podName,
+		podUID:                       podUID,
 		workflow:                     workflow,
-		workflowUID:                  workflowUID,
 		nodeId:                       nodeId,
 		ClientSet:                    clientset,
-		taskResultClient:             taskSetClient,
+		taskResultClient:             taskResultClient,
 		RESTClient:                   restClient,
 		Namespace:                    namespace,
 		RuntimeExecutor:              cre,

workflow/executor/executor_test.go

@@ -22,7 +22,7 @@ import (
 const (
 	fakePodName       = "fake-test-pod-1234567890"
 	fakeWorkflow      = "my-wf"
-	fakeWorkflowUID   = "my-wf-uid"
+	fakePodUID        = "my-pod-uid"
 	fakeNodeID        = "my-node-id"
 	fakeNamespace     = "default"
 	fakeContainerName = "main"
@@ -399,10 +399,10 @@ func TestMonitorProgress(t *testing.T) {
 		taskResults,
 		nil,
 		fakePodName,
+		fakePodUID,
 		fakeWorkflow,
 		fakeNodeID,
 		fakeNamespace,
-		fakeWorkflowUID,
 		&mocks.ContainerRuntimeExecutor{},
 		wfv1.Template{},
 		false,

workflow/executor/taskresult.go

@@ -51,10 +51,10 @@ func (we *WorkflowExecutor) createTaskResult(ctx context.Context, result wfv1.No
 	taskResult.SetOwnerReferences(
 		[]metav1.OwnerReference{
 			{
-				APIVersion: workflow.APIVersion,
-				Kind:       workflow.WorkflowKind,
-				Name:       we.workflow,
-				UID:        we.workflowUID,
+				APIVersion: "v1",
+				Kind:       "pods",
+				Name:       we.PodName,
+				UID:        we.podUID,
 			},
 		})