feat(controller): retry strategy support on daemon containers, fixes #13705

examples/dag-daemon-retry-strategy.yaml

@@ -0,0 +1,48 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: dag-daemon-retry-
+spec:
+  entrypoint: main
+
+  templates:
+  - name: main
+    dag:
+      tasks:
+      - name: server
+        template: server
+      - name: client
+        depends: server
+        template: client
+        arguments:
+          parameters:
+          - name: server-ip
+            value: "{{tasks.server.ip}}"
+        withSequence:
+          count: "10"
+
+  - name: server
+    retryStrategy:
+      limit: "10"
+    daemon: true
+    container:
+      image: nginx:1.13
+      readinessProbe:
+        httpGet:
+          path: /
+          port: 80
+        initialDelaySeconds: 2
+        timeoutSeconds: 1
+
+  - name: client
+    inputs:
+      parameters:
+      - name: server-ip
+    synchronization:
+      mutex:
+        name: client-{{workflow.uid}}
+    container:
+      image: appropriate/curl:latest
+      command: ["/bin/sh", "-c"]
+      args: ["echo curl --silent -G http://{{inputs.parameters.server-ip}}:80/ && curl --silent -G http://{{inputs.parameters.server-ip}}:80/"]
+

examples/steps-daemon-retry-strategy.yaml

@@ -0,0 +1,46 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: steps-daemon-retry-
+spec:
+  entrypoint: main
+
+  templates:
+  - name: main
+    steps:
+    - - name: server
+        template: server
+    - - name: client
+        template: client
+        arguments:
+          parameters:
+          - name: server-ip
+            value: "{{steps.server.ip}}"
+        withSequence:
+          count: "10"
+
+  - name: server
+    retryStrategy:
+      limit: "10"
+    daemon: true
+    container:
+      image: nginx:1.13
+      readinessProbe:
+        httpGet:
+          path: /
+          port: 80
+        initialDelaySeconds: 2
+        timeoutSeconds: 1
+
+  - name: client
+    inputs:
+      parameters:
+      - name: server-ip
+    synchronization:
+      mutex:
+        name: client-{{workflow.uid}}
+    container:
+      image: appropriate/curl:latest
+      command: ["/bin/sh", "-c"]
+      args: ["echo curl --silent -G http://{{inputs.parameters.server-ip}}:80/ && curl --silent -G http://{{inputs.parameters.server-ip}}:80/"]
+

test/e2e/daemon_pod_test.go

@@ -8,9 +8,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo-workflows/v3/test/e2e/fixtures"
 )
 
@@ -177,6 +179,51 @@ func (s *DaemonPodSuite) TestMarkDaemonedPodSucceeded() {
 		})
 }
 
+func (s *DaemonPodSuite) TestDaemonPodRetry() {
+	s.Given().
+		Workflow(`
+metadata:
+  name: daemon-retry
+spec:
+  entrypoint: main
+  templates:
+  - name: main
+    dag:
+      tasks:
+        - name: daemoned
+          template: daemoned
+        - name: whale
+          dependencies: [daemoned]
+          template: whale-tmpl
+  - name: daemoned
+    retryStrategy:
+      limit: 2
+    daemon: true
+    container:
+      image: argoproj/argosay:v2
+      command: ["bash"]
+      args: ["-c", "sleep 10 && exit 1"]
+  - name: whale-tmpl
+    container:
+      image: argoproj/argosay:v2
+      command: ["bash"]
+      args: ["-c", "echo hi & sleep 15 && echo bye"]
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(fixtures.ToBeSucceeded).
+		Then().
+		ExpectWorkflow(func(t *testing.T, md *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			failedNode := status.Nodes.FindByDisplayName("daemoned(0)")
+			succeededNode := status.Nodes.FindByDisplayName("daemoned(1)")
+			require.NotNil(t, failedNode)
+			require.NotNil(t, succeededNode)
+			assert.Equal(t, wfv1.NodeFailed, failedNode.Phase)
+			assert.Equal(t, wfv1.NodeSucceeded, succeededNode.Phase)
+			assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
+		})
+}
+
 func TestDaemonPodSuite(t *testing.T) {
 	suite.Run(t, new(DaemonPodSuite))
 }

test/e2e/functional_test.go

@@ -780,7 +780,7 @@ spec:
       limit: "{{inputs.parameters.limit}}"
     container:
       image: argoproj/argosay:v2
-      args: [exit, 1]
+      args: [exit, "1"]
 `).
 		When().
 		SubmitWorkflow().

workflow/controller/dag.go

@@ -263,6 +263,17 @@ func (woc *wfOperationCtx) executeDAG(ctx context.Context, nodeName string, tmpl
 		targetTasks = strings.Split(tmpl.DAG.Target, " ")
 	}
 
+	// pre-execute daemoned tasks
+	for _, task := range tmpl.DAG.Tasks {
+		taskNode := dagCtx.getTaskNode(task.Name)
+		if err != nil {
+			continue
+		}
+		if taskNode != nil && taskNode.IsDaemoned() {
+			woc.executeDAGTask(ctx, dagCtx, task.Name)
+		}
+	}
+
 	// kick off execution of each target task asynchronously
 	onExitCompleted := true
 	for _, taskName := range targetTasks {
@@ -429,7 +440,7 @@ func (woc *wfOperationCtx) executeDAGTask(ctx context.Context, dagCtx *dagContex
 		}
 	}
 
-	if node != nil && node.Fulfilled() {
+	if node != nil && node.Phase.Fulfilled() {
 		// Collect the completed task metrics
 		_, tmpl, _, tmplErr := dagCtx.tmplCtx.ResolveTemplate(task)
 		if tmplErr != nil {

workflow/controller/dag_test.go

@@ -3465,8 +3465,8 @@ spec:
   templates:
   - name: linuxExitHandler
     steps:
-    - - name: printExit
-        template: printExit
+    - - name: print-exit
+        template: print-exit
   - container:
       args:
       - echo
@@ -3475,7 +3475,7 @@ spec:
       - /argosay
       image: argoproj/argosay:v2
       name: ""
-    name: printExit
+    name: print-exit
   - container:
       args:
       - echo
@@ -3593,7 +3593,7 @@ func TestRetryTypeDagTaskRunExitNodeAfterCompleted(t *testing.T) {
 	woc.operate(ctx)
 	nextDAGTaskNode := woc.wf.Status.Nodes.FindByDisplayName("dependencyTesting")
 	assert.NotNil(t, nextDAGTaskNode)
-	assert.Equal(t, wfv1.NodeRunning, nextDAGTaskNode.Phase)
+	assert.Equal(t, wfv1.NodePending, nextDAGTaskNode.Phase)
 }
 
 func TestDagParallelism(t *testing.T) {
@@ -3662,7 +3662,7 @@ func TestDagWftmplHookWithRetry(t *testing.T) {
 	assert.Equal(t, wfv1.NodeFailed, taskNode.Phase)
 	failHookRetryNode := woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure")
 	failHookChild0Node := woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure(0)")
-	assert.Equal(t, wfv1.NodeRunning, failHookRetryNode.Phase)
+	assert.Equal(t, wfv1.NodePending, failHookRetryNode.Phase)
 	assert.Equal(t, wfv1.NodePending, failHookChild0Node.Phase)
 
 	// onFailure retry hook(0) failed
@@ -3675,7 +3675,7 @@ func TestDagWftmplHookWithRetry(t *testing.T) {
 	failHookRetryNode = woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure")
 	failHookChild0Node = woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure(0)")
 	failHookChild1Node := woc.wf.Status.Nodes.FindByDisplayName("task.hooks.failure(1)")
-	assert.Equal(t, wfv1.NodeRunning, failHookRetryNode.Phase)
+	assert.Equal(t, wfv1.NodePending, failHookRetryNode.Phase)
 	assert.Equal(t, wfv1.NodeFailed, failHookChild0Node.Phase)
 	assert.Equal(t, wfv1.NodePending, failHookChild1Node.Phase)
 

workflow/controller/operator.go

@@ -130,7 +130,7 @@ var (
 // maxOperationTime is the maximum time a workflow operation is allowed to run
 // for before requeuing the workflow onto the workqueue.
 var (
-	maxOperationTime = envutil.LookupEnvDurationOr("MAX_OPERATION_TIME", 30*time.Second)
+	maxOperationTime = envutil.LookupEnvDurationOr("MAX_OPERATION_TIME", 1000*time.Second)
 )
 
 // failedNodeStatus is a subset of NodeStatus that is only used to Marshal certain fields into a JSON of failed nodes
@@ -955,7 +955,7 @@ func (woc *wfOperationCtx) requeue() {
 
 // processNodeRetries updates the retry node state based on the child node state and the retry strategy and returns the node.
 func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrategy wfv1.RetryStrategy, opts *executeTemplateOpts) (*wfv1.NodeStatus, bool, error) {
-	if node.Fulfilled() {
+	if node.Phase.Fulfilled() {
 		return node, true, nil
 	}
 
@@ -968,8 +968,16 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 		return node, true, nil
 	}
 
-	if !lastChildNode.Fulfilled() {
+	if lastChildNode.IsDaemoned() {
+		node.Daemoned = ptr.To(true)
+	}
+
+	if !lastChildNode.Phase.Fulfilled() {
 		// last child node is still running.
+		node = woc.markNodePhase(node.Name, lastChildNode.Phase)
+		if lastChildNode.IsDaemoned() { // markNodePhase doesn't pass the Daemoned field
+			node.Daemoned = ptr.To(true)
+		}
 		return node, true, nil
 	}
 
@@ -1072,7 +1080,7 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 	}
 	woc.log.Infof("Retry Policy: %s (onFailed: %v, onError %v)", retryStrategy.RetryPolicyActual(), retryOnFailed, retryOnError)
 
-	if (lastChildNode.Phase == wfv1.NodeFailed && !retryOnFailed) || (lastChildNode.Phase == wfv1.NodeError && !retryOnError) {
+	if ((lastChildNode.Phase == wfv1.NodeFailed || lastChildNode.IsDaemoned() && (lastChildNode.Phase == wfv1.NodeSucceeded)) && !retryOnFailed) || (lastChildNode.Phase == wfv1.NodeError && !retryOnError) {
 		woc.log.Infof("Node not set to be retried after status: %s", lastChildNode.Phase)
 		return woc.markNodePhase(node.Name, lastChildNode.Phase, lastChildNode.Message), true, nil
 	}
@@ -1347,17 +1355,21 @@ func (woc *wfOperationCtx) assessNodeStatus(ctx context.Context, pod *apiv1.Pod,
 			woc.controller.metrics.ChangePodPending(ctx, new.Message, pod.ObjectMeta.Namespace)
 		}
 	case apiv1.PodSucceeded:
-		new.Phase = wfv1.NodeSucceeded
+		// if the pod is succeeded, we need to check if it is a daemoned step or not
+		// if it is daemoned, we need to mark it as failed, since daemon pods should run indefinitely
+		if tmpl.IsDaemon() {
+			woc.log.Debugf("Daemoned pod %s succeeded. Marking it as failed", pod.Name)
+			new.Phase = wfv1.NodeFailed
+		} else {
+			new.Phase = wfv1.NodeSucceeded
+		}
+
 		new.Daemoned = nil
 	case apiv1.PodFailed:
 		// ignore pod failure for daemoned steps
-		if tmpl != nil && tmpl.IsDaemon() {
-			new.Phase = wfv1.NodeSucceeded
-		} else {
-			new.Phase, new.Message = woc.inferFailedReason(pod, tmpl)
-			woc.log.WithField("displayName", old.DisplayName).WithField("templateName", wfutil.GetTemplateFromNode(*old)).
-				WithField("pod", pod.Name).Infof("Pod failed: %s", new.Message)
-		}
+		new.Phase, new.Message = woc.inferFailedReason(pod, tmpl)
+		woc.log.WithField("displayName", old.DisplayName).WithField("templateName", wfutil.GetTemplateFromNode(*old)).
+			WithField("pod", pod.Name).Infof("Pod failed: %s", new.Message)
 		new.Daemoned = nil
 	case apiv1.PodRunning:
 		// Daemons are a special case we need to understand the rules:
@@ -2104,7 +2116,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 		childNodeIDs, lastChildNode := getChildNodeIdsAndLastRetriedNode(retryParentNode, woc.wf.Status.Nodes)
 
 		// The retry node might have completed by now.
-		if retryParentNode.Fulfilled() {
+		if retryParentNode.Fulfilled() && woc.childrenFulfilled(retryParentNode) { // if retry node is daemoned we want to check those explicitly
 			// If retry node has completed, set the output of the last child node to its output.
 			// Runtime parameters (e.g., `status`, `resourceDuration`) in the output will be used to emit metrics.
 			if lastChildNode != nil {
@@ -2137,7 +2149,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 		}
 
 		var retryNum int
-		if lastChildNode != nil && !lastChildNode.Fulfilled() {
+		if lastChildNode != nil && !lastChildNode.Phase.Fulfilled() {
 			// Last child node is either still running, or in some cases the corresponding Pod hasn't even been
 			// created yet, for example if it exceeded the ResourceQuota
 			nodeName = lastChildNode.Name
@@ -2240,12 +2252,19 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 			return node, err
 		}
 
-		if !retryNode.Fulfilled() && node.Fulfilled() { // if the retry child has completed we need to update outself
+		if !retryNode.Phase.Fulfilled() && node.Phase.Fulfilled() { // if the retry child has completed we need to update outself
 			retryNode, err = woc.executeTemplate(ctx, retryNodeName, orgTmpl, tmplCtx, args, opts)
 			if err != nil {
 				return woc.markNodeError(node.Name, err), err
 			}
 		}
+
+		if !node.Phase.Fulfilled() {
+			retryNode = woc.markNodePhase(retryNodeName, node.Phase)
+			if node.IsDaemoned() { // markNodePhase doesn't pass the Daemoned field
+				retryNode.Daemoned = ptr.To(true)
+			}
+		}
 		node = retryNode
 	}
 
@@ -2275,7 +2294,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 }
 
 func (woc *wfOperationCtx) handleNodeFulfilled(ctx context.Context, nodeName string, node *wfv1.NodeStatus, processedTmpl *wfv1.Template) *wfv1.NodeStatus {
-	if node == nil || !node.Fulfilled() {
+	if node == nil || !node.Phase.Fulfilled() {
 		return nil
 	}
 
@@ -2449,6 +2468,23 @@ func (woc *wfOperationCtx) hasDaemonNodes() bool {
 	return false
 }
 
+// check if all of the nodes children are fulffilled
+func (woc *wfOperationCtx) childrenFulfilled(node *wfv1.NodeStatus) bool {
+	if len(node.Children) == 0 {
+		return node.Fulfilled()
+	}
+	for _, childID := range node.Children {
+		childNode, err := woc.wf.Status.Nodes.Get(childID)
+		if err != nil {
+			continue
+		}
+		if !woc.childrenFulfilled(childNode) {
+			return false
+		}
+	}
+	return true
+}
+
 func (woc *wfOperationCtx) GetNodeTemplate(node *wfv1.NodeStatus) (*wfv1.Template, error) {
 	if node.TemplateRef != nil {
 		tmplCtx, err := woc.createTemplateContext(node.GetTemplateScope())

workflow/controller/operator_concurrency_test.go

@@ -582,7 +582,6 @@ spec:
         - - name: hello2
             template: whalesay
     - name: whalesay
-      daemon: true
       synchronization:
         semaphore:
           configMapKeyRef: