Repository: argoproj/argo (go)

Issue:
Exceeding quota with volumeClaimTemplates
Checklist:

 * [x] I've included the version.
 * [x] I've included reproduction steps.
 * [x] I've included the workflow YAML.
 * [x] I've included the logs.
 
**What happened**:
I have created a template which uses a volume claim template. If the total claimed by running pods is at storage quota then new pods being attempted to run enter error state rather than pending state.

**What you expected to happen**:
If storage resources are unavailable then workflow should remain pending until resources are available.

**How to reproduce it (as minimally and precisely as possible)**:
In a namespace with a quota of 10Gi of storage the error should start occuring after 5 have been submitted.
```
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: storage-limit-
spec:
  serviceAccountName: argo
  entrypoint: wait
  podGC:
    strategy: OnPodCompletion

  volumeClaimTemplates:                 # define volume, same syntax as k8s Pod spec
    - metadata:
        name: workdir                     # name of volume claim
      spec:
        accessModes: [ "ReadWriteMany" ]
        resources:
          requests:
            storage: 2Gi                  # Gi => 1024 * 1024 * 1024

  templates:
  - name: wait
    resubmitPendingPods: true
    script:
      image: alpine:latest
      command: [sh, -c]
      args: ["sleep", "30s"]
```
Use bash script to run this some number of times until you encounter the issue.

**Anything else we need to know?**:
I believe this issue also occurs when hitting PVC limits.

**Environment**:

- Argo version:
```
$ argo version
argo: 2.8.2+8a151ae.dirty
  BuildDate: 2020-06-18T23:50:58Z
  GitCommit: 8a151aec6538c9442cf2380c2544ba3efb60ff60
  GitTreeState: dirty
  GitTag: 2.8.2
  GoVersion: go1.13
  Compiler: gc
  Platform: linux/amd64
```
- Kubernetes version :
```
$ kubectl version -o yaml
clientVersion:
  buildDate: 2020-01-29T21:26:39Z
  compiler: gc
  gitCommit: d4cacc0
  gitTreeState: clean
  gitVersion: v1.10.0+d4cacc0
  goVersion: go1.14beta1
  major: "1"
  minor: 10+
  platform: linux/amd64
serverVersion:
  buildDate: 2020-05-04T12:54:43Z
  compiler: gc
  gitCommit: a3ec9df
  gitTreeState: clean
  gitVersion: v1.16.2
  goVersion: go1.12.12
  major: "1"
  minor: 16+
  platform: linux/amd64
```

**Other debugging information (if applicable)**:

- workflow result:

```
$ argo --loglevel DEBUG get <workflowname>
DEBU[0000] CLI version                                   version="{2.8.2+8a151ae.dirty 2020-06-18T23:50:58Z 8a151aec6538c9442cf2380c2544ba3efb60ff60 2.8.2 dirty go1.13 gc linux/amd64}"
DEBU[0000] Client options                                opts="{{ false false} 0x1574670 0xc00011f040}"
Name:                storage-limit-v9qjt
Namespace:           thoth-test-core
ServiceAccount:      argo
Status:              Error
Message:             persistentvolumeclaims "storage-limit-v9qjt-workdir" is forbidden: exceeded quota: thoth-test-core-quota, requested: requests.storage=2Gi, used: requests.storage=40Gi, limited: requests.storage=40Gi
Conditions:          
 Completed           True
Created:             Wed Jul 08 11:57:57 -0400 (7 minutes ago)
Started:             Wed Jul 08 11:57:57 -0400 (7 minutes ago)
Finished:            Wed Jul 08 11:57:57 -0400 (7 minutes ago)
Duration:            0 seconds
```

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you are impacted by this bug please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
fix: Exceeding quota with volumeClaimTemplates
Checklist:

* [ ] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [ ] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  Fixes #3419 
* [ ] I've signed the CLA.
* [ ] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [ ] My builds are green. Try syncing with master if they are not. 
* [ ] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).


Patch:
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 89be36f0c..1fb929d3d 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -282,11 +282,21 @@ func (woc *wfOperationCtx) operate() {
 
 	err = woc.createPVCs()
 	if err != nil {
+		if apierr.IsForbidden(err) {
+			// Error was most likely caused by a lack of resources.
+			// In this case, Workflow will be in pending state and requeue.
+			woc.markWorkflowPhase(wfv1.NodePending, false, fmt.Sprintf("Waiting for a PVC to be created. %v", err))
+			woc.requeue(10)
+			return
+		}
 		msg := "pvc create error"
 		woc.log.WithError(err).Error(msg)
 		woc.markWorkflowError(err, true)
 		woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", fmt.Sprintf("%s %s: %+v", woc.wf.ObjectMeta.Name, msg, err))
 		return
+	} else if woc.wf.Status.Phase == wfv1.NodePending {
+		// Workflow might be in pending state if previous PVC creation is forbidden
+		woc.markWorkflowRunning()
 	}
 
 	node, err := woc.executeTemplate(woc.wf.ObjectMeta.Name, execTmplRef, tmplCtx, execArgs, &executeTemplateOpts{})
@@ -1226,8 +1236,8 @@ func inferFailedReason(pod *apiv1.Pod) (wfv1.NodePhase, string) {
 }
 
 func (woc *wfOperationCtx) createPVCs() error {
-	if woc.wf.Status.Phase != wfv1.NodeRunning {
-		// Only attempt to create PVCs if workflow transitioned to Running state
+	if !(woc.wf.Status.Phase == wfv1.NodePending || woc.wf.Status.Phase == wfv1.NodeRunning) {
+		// Only attempt to create PVCs if workflow is in Pending or Running state
 		// (e.g. passed validation, or didn't already complete)
 		return nil
 	}
@@ -1236,9 +1246,6 @@ func (woc *wfOperationCtx) createPVCs() error {
 		// This will also handle the case where workflow has no volumeClaimTemplates.
 		return nil
 	}
-	if len(woc.wf.Status.PersistentVolumeClaims) == 0 {
-		woc.wf.Status.PersistentVolumeClaims = make([]apiv1.Volume, len(woc.wfSpec.VolumeClaimTemplates))
-	}
 	pvcClient := woc.controller.kubeclientset.CoreV1().PersistentVolumeClaims(woc.wf.ObjectMeta.Namespace)
 	for i, pvcTmpl := range woc.wfSpec.VolumeClaimTemplates {
 		if pvcTmpl.ObjectMeta.Name == "" {
@@ -1290,7 +1297,7 @@ func (woc *wfOperationCtx) createPVCs() error {
 				},
 			},
 		}
-		woc.wf.Status.PersistentVolumeClaims[i] = vol
+		woc.wf.Status.PersistentVolumeClaims = append(woc.wf.Status.PersistentVolumeClaims, vol)
 		woc.updated = true
 	}
 	return nil
@@ -1673,7 +1680,7 @@ func (woc *wfOperationCtx) hasDaemonNodes() bool {
 }
 
 func (woc *wfOperationCtx) markWorkflowRunning() {
-	woc.markWorkflowPhase(wfv1.NodeRunning, false)
+	woc.markWorkflowPhase(wfv1.NodeRunning, false, "")
 }
 
 func (woc *wfOperationCtx) markWorkflowSuccess() {
diff --git a/workflow/util/util.go b/workflow/util/util.go
index 793bdcc36..a092431b1 100644
--- a/workflow/util/util.go
+++ b/workflow/util/util.go
@@ -566,7 +566,7 @@ func FormulateResubmitWorkflow(wf *wfv1.Workflow, memoized bool) (*wfv1.Workflow
 	}
 
 	newWF.Status.Conditions = wfv1.Conditions{{Status: metav1.ConditionFalse, Type: wfv1.ConditionTypeCompleted}}
-	newWF.Status.Phase = wfv1.NodePending
+	newWF.Status.Phase = ""
 
 	return &newWF, nil
 }


Test patch:
diff --git a/test/e2e/fixtures/when.go b/test/e2e/fixtures/when.go
index c51cf9045..bceef8536 100644
--- a/test/e2e/fixtures/when.go
+++ b/test/e2e/fixtures/when.go
@@ -35,6 +35,7 @@ type When struct {
 	cronWorkflowName  string
 	kubeClient        kubernetes.Interface
 	resourceQuota     *corev1.ResourceQuota
+	storageQuota      *corev1.ResourceQuota
 	configMap         *corev1.ConfigMap
 }
 
@@ -227,6 +228,24 @@ func (w *When) MemoryQuota(quota string) *When {
 	return w
 }
 
+func (w *When) StorageQuota(quota string) *When {
+	obj, err := util.CreateHardStorageQuota(w.kubeClient, "argo", "storage-quota", quota)
+	if err != nil {
+		w.t.Fatal(err)
+	}
+	w.storageQuota = obj
+	return w
+}
+
+func (w *When) DeleteStorageQuota() *When {
+	err := util.DeleteQuota(w.kubeClient, w.storageQuota)
+	if err != nil {
+		w.t.Fatal(err)
+	}
+	w.storageQuota = nil
+	return w
+}
+
 func (w *When) DeleteQuota() *When {
 	err := util.DeleteQuota(w.kubeClient, w.resourceQuota)
 	if err != nil {
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index 9fc50e112..460479a78 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -4,6 +4,7 @@ package e2e
 
 import (
 	"regexp"
+	"strings"
 	"testing"
 	"time"
 
@@ -639,6 +640,24 @@ spec:
 		})
 }
 
+func (s *FunctionalSuite) TestStorageQuotaLimit() {
+	s.Given().
+		Workflow("@testdata/storage-limit.yaml").
+		When().
+		StorageQuota("5Mi").
+		SubmitWorkflow().
+		WaitForWorkflowToStart(5*time.Second).
+		WaitForWorkflowCondition(func(wf *wfv1.Workflow) bool {
+			return strings.Contains(wf.Status.Message, "Waiting for a PVC to be created")
+		}, "PVC pending", 10*time.Second).
+		DeleteStorageQuota().
+		WaitForWorkflow(30 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
 func TestFunctionalSuite(t *testing.T) {
 	suite.Run(t, new(FunctionalSuite))
 }
diff --git a/test/e2e/testdata/storage-limit.yaml b/test/e2e/testdata/storage-limit.yaml
new file mode 100644
index 000000000..51b1e7964
--- /dev/null
+++ b/test/e2e/testdata/storage-limit.yaml
@@ -0,0 +1,22 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: storage-quota-limit
+  labels:
+    argo-e2e: true
+spec:
+  entrypoint: wait
+  volumeClaimTemplates:                 # define volume, same syntax as k8s Pod spec
+    - metadata:
+        name: workdir1                     # name of volume claim
+      spec:
+        accessModes: [ "ReadWriteMany" ]
+        resources:
+          requests:
+            storage: 20Mi
+
+  templates:
+  - name: wait
+    script:
+      image: argoproj/argosay:v2
+      args: [echo, ":) Hello Argo!"]
diff --git a/test/util/resourcequota.go b/test/util/resourcequota.go
index e6343ddf0..63874b8c1 100644
--- a/test/util/resourcequota.go
+++ b/test/util/resourcequota.go
@@ -8,15 +8,27 @@ import (
 )
 
 func CreateHardMemoryQuota(clientset kubernetes.Interface, namespace, name, memoryLimit string) (*corev1.ResourceQuota, error) {
+	resourceList := corev1.ResourceList{
+		corev1.ResourceLimitsMemory: resource.MustParse(memoryLimit),
+	}
+	return CreateResourceQuota(clientset, namespace, name, resourceList)
+}
+
+func CreateHardStorageQuota(clientset kubernetes.Interface, namespace, name, storageLimit string) (*corev1.ResourceQuota, error) {
+	resourceList := corev1.ResourceList{
+		"requests.storage": resource.MustParse(storageLimit),
+	}
+	return CreateResourceQuota(clientset, namespace, name, resourceList)
+}
+
+func CreateResourceQuota(clientset kubernetes.Interface, namespace, name string, rl corev1.ResourceList) (*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),
-			},
+			Hard: rl,
 		},
 	})
 }
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 95e79813d..fa49d45d1 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -4101,7 +4101,7 @@ status:
 		defer cancel()
 		woc := newWorkflowOperationCtx(wf, controller)
 		woc.operate()
-		assert.Equal(t, wfv1.NodePending, woc.wf.Status.Phase)
+		assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
 		for _, node := range woc.wf.Status.Nodes {
 			switch node.TemplateName {
 			case "main":