feat: Retry pending nodes

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -987,11 +987,16 @@ func (in *WorkflowStatus) AnyActiveSuspendNode() bool {
  return in.Nodes.Any(func(node NodeStatus) bool { return node.IsActiveSuspendNode() })
 }
 
-// Remove returns whether or not the node has completed execution
+// Completed returns whether or not the node has completed execution
 func (n NodeStatus) Completed() bool {
  return isCompletedPhase(n.Phase) || n.IsDaemoned() && n.Phase != NodePending
 }
 
+// Pending returns whether or not the node has completed execution
+func (n NodeStatus) Pending() bool {
+ return n.Phase == NodePending
+}
+
 // IsDaemoned returns whether or not the node is deamoned
 func (n NodeStatus) IsDaemoned() bool {
  if n.Daemoned == nil || !*n.Daemoned {

test/e2e/fixtures/when.go

@@ -9,12 +9,14 @@ import (
  "k8s.io/client-go/kubernetes"
 
  log "github.com/sirupsen/logrus"
+ corev1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/fields"
 
  "github.com/argoproj/argo/persist/sqldb"
  wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
  "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
+ "github.com/argoproj/argo/test/util"
  "github.com/argoproj/argo/workflow/packer"
 )
 
@@ -32,6 +34,7 @@ type When struct {
  wfTemplateNames []string
  cronWorkflowName string
  kubeClient kubernetes.Interface
+ resourceQuota *corev1.ResourceQuota
 }
 
 func (w *When) SubmitWorkflow() *When {
@@ -173,6 +176,24 @@ func (w *When) RunCli(args []string, block func(t *testing.T, output string, err
  return w
 }
 
+func (w *When) MemoryQuota(quota string) *When {
+ obj, err := util.CreateHardMemoryQuota(w.kubeClient, "argo", "memory-quota", quota)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ w.resourceQuota = obj
+ return w
+}
+
+func (w *When) DeleteQuota() *When {
+ err := util.DeleteQuota(w.kubeClient, w.resourceQuota)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ w.resourceQuota = nil
+ return w
+}
+
 func (w *When) Then() *Then {
  return &Then{
  t: w.t,

test/e2e/functional_test.go

@@ -1,6 +1,7 @@
 package e2e
 
 import (
+ "regexp"
  "strings"
  "testing"
  "time"
@@ -20,6 +21,11 @@ type FunctionalSuite struct {
  fixtures.E2ESuite
 }
 
+func (s *FunctionalSuite) TearDownSuite() {
+ s.E2ESuite.DeleteResources(fixtures.Label)
+ s.Persistence.Close()
+}
+
 func (s *FunctionalSuite) TestContinueOnFail() {
  s.Given().
  Workflow(`
@@ -178,6 +184,107 @@ func (s *FunctionalSuite) TestLoopEmptyParam() {
  })
 }
 
+// 128M is for argo executor
+func (s *FunctionalSuite) TestPendingWorkflow() {
+ s.Given().
+ Workflow(`
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ name: dag-limited-noretry
+ labels:
+ argo-e2e: true
+spec:
+ entrypoint: dag
+ templates:
+ - name: cowsay
+ container:
+ image: cowsay:v1
+ command: [sh, -c]
+ args: ["cowsay a"]
+ resources:
+ limits:
+ memory: 128M
+ - name: dag
+ dag:
+ tasks:
+ - name: a
+ template: cowsay
+ - name: b
+ template: cowsay
+`).
+ When().
+ MemoryQuota("130M").
+ SubmitWorkflow().
+ WaitForWorkflowToStart(5*time.Second).
+ WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+ a := wf.Status.Nodes.FindByDisplayName("a")
+ b := wf.Status.Nodes.FindByDisplayName("b")
+ return wfv1.NodePending == a.Phase &&
+ regexp.MustCompile(`^Pending \d+\.\d+s$`).MatchString(a.Message) &&
+ wfv1.NodePending == b.Phase &&
+ regexp.MustCompile(`^Pending \d+\.\d+s$`).MatchString(b.Message)
+ }, "pods pending", 10*time.Second).
+ DeleteQuota().
+ WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+ a := wf.Status.Nodes.FindByDisplayName("a")
+ b := wf.Status.Nodes.FindByDisplayName("b")
+ return wfv1.NodeSucceeded == a.Phase && wfv1.NodeSucceeded == b.Phase
+ }, "pods succeeded", 10*time.Second)
+ s.TearDownSuite()
+}
+
+func (s *FunctionalSuite) TestPendingRetryWorkflow() {
+ s.Given().
+ Workflow(`
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ name: dag-limited
+ labels:
+ argo-e2e: true
+spec:
+ entrypoint: dag
+ templates:
+ - name: cowsay
+ retryStrategy:
+ limit: 1
+ container:
+ image: cowsay:v1
+ command: [sh, -c]
+ args: ["cowsay a"]
+ resources:
+ limits:
+ memory: 128M
+ - name: dag
+ dag:
+ tasks:
+ - name: a
+ template: cowsay
+ - name: b
+ template: cowsay
+`).
+ When().
+ MemoryQuota("130M").
+ SubmitWorkflow().
+ WaitForWorkflowToStart(5*time.Second).
+ WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+ a := wf.Status.Nodes.FindByDisplayName("a(0)")
+ b := wf.Status.Nodes.FindByDisplayName("b(0)")
+ return wfv1.NodePending == a.Phase &&
+ regexp.MustCompile(`^Pending \d+\.\d+s$`).MatchString(a.Message) &&
+ wfv1.NodePending == b.Phase &&
+ regexp.MustCompile(`^Pending \d+\.\d+s$`).MatchString(b.Message)
+ }, "pods pending", 10*time.Second).
+ DeleteQuota().
+ WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+ a := wf.Status.Nodes.FindByDisplayName("a(0)")
+ b := wf.Status.Nodes.FindByDisplayName("b(0)")
+ return wfv1.NodeSucceeded == a.Phase && wfv1.NodeSucceeded == b.Phase
+ }, "pods succeeded", 10*time.Second)
+ s.TearDownSuite()
+}
+
 func (s *FunctionalSuite) TestparameterAggregation() {
  s.Given().
  Workflow("@functional/param-aggregation.yaml").

test/e2e/manifests/mysql.yaml

@@ -267,6 +267,15 @@ subjects:
 apiVersion: v1
 data:
  config: |
+ executor:
+ imagePullPolicy: IfNotPresent
+ resources:
+ requests:
+ cpu: 0.1
+ memory: 64Mi
+ limits:
+ cpu: 0.5
+ memory: 128Mi
  artifactRepository:
  archiveLogs: true
  s3:

test/e2e/manifests/no-db.yaml

@@ -267,6 +267,15 @@ subjects:
 apiVersion: v1
 data:
  config: |
+ executor:
+ imagePullPolicy: IfNotPresent
+ resources:
+ requests:
+ cpu: 0.1
+ memory: 64Mi
+ limits:
+ cpu: 0.5
+ memory: 128Mi
  artifactRepository:
  archiveLogs: true
  s3:

test/e2e/manifests/postgres.yaml

@@ -267,6 +267,15 @@ subjects:
 apiVersion: v1
 data:
  config: |
+ executor:
+ imagePullPolicy: IfNotPresent
+ resources:
+ requests:
+ cpu: 0.1
+ memory: 64Mi
+ limits:
+ cpu: 0.5
+ memory: 128Mi
  artifactRepository:
  archiveLogs: true
  s3:

test/util/resourcequota.go

@@ -0,0 +1,23 @@
+package util
+
+import (
+ corev1 "k8s.io/api/core/v1"
+ "k8s.io/apimachinery/pkg/api/resource"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/client-go/kubernetes"
+)
+
+func CreateHardMemoryQuota(clientset kubernetes.Interface, namespace, name, memoryLimit string) (*corev1.ResourceQuota, error) {
+ return clientset.CoreV1().ResourceQuotas(namespace).Create(&corev1.ResourceQuota{
+ ObjectMeta: metav1.ObjectMeta{Name: name},
+ Spec: corev1.ResourceQuotaSpec{
+ Hard: corev1.ResourceList{
+ corev1.ResourceLimitsMemory: resource.MustParse(memoryLimit),
+ },
+ },
+ })
+}
+
+func DeleteQuota(clientset kubernetes.Interface, quota *corev1.ResourceQuota) error {
+ return clientset.CoreV1().ResourceQuotas(quota.Namespace).Delete(quota.Name, &metav1.DeleteOptions{})
+}

workflow/controller/operator.go

@@ -748,7 +748,7 @@ func (woc *wfOperationCtx) podReconciliation() error {
  // It is now impossible to infer pod status. The only thing we can do at this point is to mark
  // the node with Error.
  for nodeID, node := range woc.wf.Status.Nodes {
- if node.Type != wfv1.NodeTypePod || node.Completed() || node.StartedAt.IsZero() {
+ if node.Type != wfv1.NodeTypePod || node.Completed() || node.StartedAt.IsZero() || node.Pending() {
  // node is not a pod, it is already complete, or it can be re-run.
  continue
  }
@@ -1338,6 +1338,16 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
  if err != nil {
  return woc.markNodeError(retryNodeName, err), err
  }
+ if lastChildNode != nil && lastChildNode.Pending() && processedTmpl.GetType() == wfv1.TemplateTypeContainer {
+ _, err := woc.createWorkflowPod(lastChildNode.Name, *processedTmpl.Container, processedTmpl, false)
+ if apierr.IsForbidden(err) {
+ return woc.markNodePending(lastChildNode.Name, err), nil
+ }
+ if err != nil {
+ return woc.markNodeError(lastChildNode.Name, err), err
+ }
+ return woc.markNodePhase(lastChildNode.Name, wfv1.NodeRunning), nil
+ }
  if lastChildNode != nil && !lastChildNode.Completed() {
  // Last child node is still running.
  nodeName = lastChildNode.Name
@@ -1375,6 +1385,9 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
  err = errors.Errorf(errors.CodeBadRequest, "Template '%s' missing specification", processedTmpl.Name)
  return woc.initializeNode(nodeName, wfv1.NodeTypeSkipped, templateScope, orgTmpl, boundaryID, wfv1.NodeError, err.Error()), err
  }
+ if apierr.IsForbidden(err) {
+ return woc.markNodePending(node.Name, err), nil
+ }
  if err != nil {
  node = woc.markNodeError(node.Name, err)
  // If retry policy is not set, or if it is not set to Always or OnError, we won't attempt to retry an errored container
@@ -1592,6 +1605,13 @@ func (woc *wfOperationCtx) markNodeError(nodeName string, err error) *wfv1.NodeS
  return woc.markNodePhase(nodeName, wfv1.NodeError, err.Error())
 }
 
+// markNodePending is a convenience method to mark a node and set the message from the error
+func (woc *wfOperationCtx) markNodePending(nodeName string, err error) *wfv1.NodeStatus {
+ woc.log.Infof("Mark node %s as Pending, due to: %+v", nodeName, err)
+ node := woc.getNodeByName(nodeName)
+ return woc.markNodePhase(nodeName, wfv1.NodePending, fmt.Sprintf("Pending %s", time.Since(node.StartedAt.Time)))
+}
+
 // checkParallelism checks if the given template is able to be executed, considering the current active pods and workflow/template parallelism
 func (woc *wfOperationCtx) checkParallelism(tmpl *wfv1.Template, node *wfv1.NodeStatus, boundaryID string) error {
  if woc.wf.Spec.Parallelism != nil && woc.activePods >= *woc.wf.Spec.Parallelism {
@@ -1638,6 +1658,10 @@ func (woc *wfOperationCtx) checkParallelism(tmpl *wfv1.Template, node *wfv1.Node
 func (woc *wfOperationCtx) executeContainer(nodeName string, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateHolder, boundaryID string) (*wfv1.NodeStatus, error) {
  node := woc.getNodeByName(nodeName)
  if node != nil {
+ if node.Pending() {
+ _, err := woc.createWorkflowPod(node.Name, *tmpl.Container, tmpl, false)
+ return node, err
+ }
  return node, nil
  }
  node = woc.initializeExecutableNode(nodeName, wfv1.NodeTypePod, templateScope, tmpl, orgTmpl, boundaryID, wfv1.NodePending)

workflow/controller/workflowpod.go

@@ -292,6 +292,9 @@ func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Cont
  woc.log.Infof("Skipped pod %s (%s) creation: already exists", nodeName, nodeID)
  return created, nil
  }
+ if apierr.IsForbidden(err) {
+ return nil, err
+ }
  woc.log.Infof("Failed to create pod %s (%s): %v", nodeName, nodeID, err)
  return nil, errors.InternalWrapError(err)
  }