feat: Retry pending nodes

api/openapi-spec/swagger.json

@@ -2208,6 +2208,11 @@
           "format": "int32",
           "title": "Limit is the maximum number of attempts when retrying a container"
         },
+        "resubmit": {
+          "type": "boolean",
+          "format": "boolean",
+          "title": "Resubmit is set to enable Pending pod resubmission"
+        },
         "retryPolicy": {
           "type": "string",
           "title": "RetryPolicy is a policy of NodePhase statuses that will be retried"

manifests/quick-start/mysql/overlays/workflow-controller-configmap.yaml

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

manifests/quick-start/no-db/overlays/workflow-controller-configmap.yaml

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

manifests/quick-start/postgres/overlays/workflow-controller-configmap.yaml

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

pkg/apiclient/cronworkflow/cron-workflow.swagger.json

@@ -1095,6 +1095,11 @@
         "backoff": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.Backoff",
           "title": "Backoff is a backoff strategy"
+        },
+        "resubmit": {
+          "type": "boolean",
+          "format": "boolean",
+          "title": "Resubmit is set to enable Pending pod resubmission"
         }
       },
       "title": "RetryStrategy provides controls on how to retry a workflow step"

pkg/apiclient/workflow/workflow.swagger.json

@@ -1419,6 +1419,11 @@
         "backoff": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.Backoff",
           "title": "Backoff is a backoff strategy"
+        },
+        "resubmit": {
+          "type": "boolean",
+          "format": "boolean",
+          "title": "Resubmit is set to enable Pending pod resubmission"
         }
       },
       "title": "RetryStrategy provides controls on how to retry a workflow step"

pkg/apiclient/workflowarchive/workflow-archive.swagger.json

@@ -888,6 +888,11 @@
         "backoff": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.Backoff",
           "title": "Backoff is a backoff strategy"
+        },
+        "resubmit": {
+          "type": "boolean",
+          "format": "boolean",
+          "title": "Resubmit is set to enable Pending pod resubmission"
         }
       },
       "title": "RetryStrategy provides controls on how to retry a workflow step"

pkg/apiclient/workflowtemplate/workflow-template.swagger.json

@@ -947,6 +947,11 @@
         "backoff": {
           "$ref": "#/definitions/github.com.argoproj.argo.pkg.apis.workflow.v1alpha1.Backoff",
           "title": "Backoff is a backoff strategy"
+        },
+        "resubmit": {
+          "type": "boolean",
+          "format": "boolean",
+          "title": "Resubmit is set to enable Pending pod resubmission"
         }
       },
       "title": "RetryStrategy provides controls on how to retry a workflow step"

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -875,6 +875,9 @@ type RetryStrategy struct {
 
 	// Backoff is a backoff strategy
 	Backoff *Backoff `json:"backoff,omitempty" protobuf:"bytes,3,opt,name=backoff,casttype=Backoff"`
+
+	// Resubmit is set to enable Pending pod resubmission
+	Resubmit *bool `json:"resubmit,omitempty" protobuf:"varint,4,opt,name=resubmit"`
 }
 
 // NodeStatus contains status information about an individual node in the workflow
@@ -987,11 +990,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 is in pending state
+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,59 @@ func (s *FunctionalSuite) TestLoopEmptyParam() {
 		})
 }
 
+// 128M is for argo executor
+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
+      resubmit: true
+    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", 20*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", 20*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,29 @@
+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,
+			Labels: map[string]string{"argo-e2e": "true"},
+		},
+		Spec: corev1.ResourceQuotaSpec{
+			Hard: corev1.ResourceList{
+				corev1.ResourceLimitsMemory: resource.MustParse(memoryLimit),
+			},
+		},
+	})
+}
+
+func DeleteQuota(clientset kubernetes.Interface, quota *corev1.ResourceQuota) error {
+	if quota == nil {
+		return nil
+	}
+	return clientset.CoreV1().ResourceQuotas(quota.Namespace).Delete(quota.Name, &metav1.DeleteOptions{})
+}