fix: Script steps fail with exceededQuota

workflow/controller/operator.go

@@ -1864,11 +1864,10 @@ func (woc *wfOperationCtx) executeContainer(nodeName string, templateScope strin
 		executionDeadline:   opts.executionDeadline,
 	})
 
-	if apierr.IsForbidden(err) && isResubmitAllowed(tmpl) {
-		// Our error was most likely caused by a lack of resources. If pod resubmission is allowed, keep the node pending
-		woc.requeue(0)
-		return woc.markNodePending(node.Name, err), nil
+	if err != nil {
+		return woc.checkForbiddenErrorAndResbmitAllowed(err, node.Name, tmpl)
 	}
+
 	return node, err
 }
 
@@ -2003,9 +2002,21 @@ func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string,
 		onExitPod:           opts.onExitTemplate,
 		executionDeadline:   opts.executionDeadline,
 	})
+	if err != nil {
+		return woc.checkForbiddenErrorAndResbmitAllowed(err, node.Name, tmpl)
+	}
 	return node, err
 }
 
+func (woc *wfOperationCtx) checkForbiddenErrorAndResbmitAllowed(err error, nodeName string, tmpl *wfv1.Template) (*wfv1.NodeStatus, error) {
+	if apierr.IsForbidden(err) && isResubmitAllowed(tmpl) {
+		// Our error was most likely caused by a lack of resources. If pod resubmission is allowed, keep the node pending
+		woc.requeue(0)
+		return woc.markNodePending(nodeName, err), nil
+	}
+	return nil, err
+}
+
 // buildLocalScope adds all of a nodes outputs to the local scope with the given prefix, as well
 // as the global scope, if specified with a globalName
 func (woc *wfOperationCtx) buildLocalScope(scope *wfScope, prefix string, node *wfv1.NodeStatus) {
@@ -2264,6 +2275,10 @@ func (woc *wfOperationCtx) executeResource(nodeName string, templateScope string
 	mainCtr := woc.newExecContainer(common.MainContainerName, tmpl)
 	mainCtr.Command = []string{"argoexec", "resource", tmpl.Resource.Action}
 	_, err = woc.createWorkflowPod(nodeName, *mainCtr, tmpl, &createWorkflowPodOpts{onExitPod: opts.onExitTemplate, executionDeadline: opts.executionDeadline})
+	if err != nil {
+		return woc.checkForbiddenErrorAndResbmitAllowed(err, node.Name, tmpl)
+	}
+
 	return node, err
 }
 

workflow/controller/operator_test.go

@@ -11,8 +11,10 @@ import (
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
 	apiv1 "k8s.io/api/core/v1"
+	apierr "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"sigs.k8s.io/yaml"
 
@@ -3902,3 +3904,79 @@ func TestPropagateMaxDurationProcess(t *testing.T) {
 		assert.Equal(t, n.StartedAt.Add(20*time.Second).Round(time.Second).String(), opts.executionDeadline.Round(time.Second).String())
 	}
 }
+
+var resubmitPendingWf = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  creationTimestamp: "2020-07-07T19:54:18Z"
+  generation: 2
+  labels:
+    workflows.argoproj.io/phase: Running
+  name: resubmit-pending-wf
+  namespace: argo
+  resourceVersion: "267165"
+  selfLink: /apis/argoproj.io/v1alpha1/namespaces/argo/workflows/resubmit-pending-wf
+  uid: 19bc715f-de50-41e6-9ae7-a54e8e0a9fb6
+spec:
+  arguments: {}
+  entrypoint: resubmit-pending
+  templates:
+  - arguments: {}
+    inputs: {}
+    metadata: {}
+    name: resubmit-pending
+    outputs: {}
+    resubmitPendingPods: true
+    script:
+      command:
+      - bash
+      image: busybox
+      name: ""
+      resources:
+        limits:
+          cpu: "10"
+      source: |
+        sleep 5
+status:
+  finishedAt: null
+  nodes:
+    resubmit-pending-wf:
+      displayName: resubmit-pending-wf
+      finishedAt: null
+      id: resubmit-pending-wf
+      message: Pending 156.62ms
+      name: resubmit-pending-wf
+      phase: Pending
+      startedAt: "2020-07-07T19:54:18Z"
+      templateName: resubmit-pending
+      templateScope: local/resubmit-pending-wf
+      type: Pod
+  phase: Running
+  startedAt: "2020-07-07T19:54:18Z"
+`
+
+func TestCheckForbiddenErrorAndResbmitAllowed(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	assert.NotNil(t, controller)
+	wf := unmarshalWF(resubmitPendingWf)
+	assert.NotNil(t, wf)
+	woc := newWorkflowOperationCtx(wf, controller)
+	assert.NotNil(t, woc)
+
+	forbiddenErr := apierr.NewForbidden(schema.GroupResource{Group: "test", Resource: "test1"}, "test", nil)
+	nonForbiddenErr := apierr.NewBadRequest("badrequest")
+	t.Run("ForbiddenError", func(t *testing.T) {
+		node, err := woc.checkForbiddenErrorAndResbmitAllowed(forbiddenErr, "resubmit-pending-wf", &wf.Spec.Templates[0])
+		assert.NotNil(t, node)
+		assert.Equal(t, wfv1.NodePending, node.Phase)
+		assert.Nil(t, err)
+	})
+	t.Run("NonForbiddenError", func(t *testing.T) {
+		node, err := woc.checkForbiddenErrorAndResbmitAllowed(nonForbiddenErr, "resubmit-pending-wf", &wf.Spec.Templates[0])
+		assert.Nil(t, node)
+		assert.NotNil(t, err)
+	})
+
+}