Repository: argoproj/argo (go)

Issue:
Argo Workflows does not honour global timeout if step/pod is not able to schedule 
Checklist:

 I've included reproduction steps.

**What happened**:

Global timeout is set to 6 hours for workflow. Because cluster was not scaled one the step in workflow was not able to schedule. 

**What you expected to happen**:

As global timeout is set to 6 hours. Workflow should timeout instead of keeping it running. 

**How to reproduce it (as minimally and precisely as possible)**:

Submit a workflow where one the step is requesting higher resources than it will be able to schedule. 

Example: If nodes in cluster are c5.xlarge. Try scheduling step with 5 cpus. Workflow step should in scheduling state and wont timeout. 

Let me know if you need more information.

Pull request description:
fix: Argo Workflows does not honour global timeout if step/pod is not able to schedule
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 #3571
* [ ] 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 75f820949..e787e1340 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -224,6 +224,12 @@ func (woc *wfOperationCtx) operate() {
 
 		woc.workflowDeadline = woc.getWorkflowDeadline()
 
+		// Workflow will not be requeued if workflow steps are in pending state.
+		// Workflow needs to requeue on its deadline,
+		if woc.workflowDeadline != nil {
+			woc.requeue(time.Until(*woc.workflowDeadline))
+		}
+
 		if woc.wfSpec.Metrics != nil {
 			realTimeScope := map[string]func() float64{common.GlobalVarWorkflowDuration: func() float64 {
 				return time.Since(woc.wf.Status.StartedAt.Time).Seconds()
@@ -234,7 +240,7 @@ func (woc *wfOperationCtx) operate() {
 		woc.workflowDeadline = woc.getWorkflowDeadline()
 		err := woc.podReconciliation()
 		if err == nil {
-			err = woc.failSuspendedNodesAfterDeadlineOrShutdown()
+			err = woc.failSuspendedAndPendingNodesAfterDeadlineOrShutdown()
 		}
 		if err != nil {
 			woc.log.WithError(err).WithField("workflow", woc.wf.ObjectMeta.Name).Error("workflow timeout")
@@ -881,11 +887,12 @@ func (woc *wfOperationCtx) shouldPrintPodSpec(node wfv1.NodeStatus) bool {
 		(woc.controller.Config.PodSpecLogStrategy.FailedPod && node.FailedOrError())
 }
 
-//fails any suspended nodes if the workflow deadline has passed
-func (woc *wfOperationCtx) failSuspendedNodesAfterDeadlineOrShutdown() error {
-	if woc.wfSpec.Shutdown != "" || (woc.workflowDeadline != nil && time.Now().UTC().After(*woc.workflowDeadline)) {
+//fails any suspended and pending nodes if the workflow deadline has passed
+func (woc *wfOperationCtx) failSuspendedAndPendingNodesAfterDeadlineOrShutdown() error {
+	deadlineExceeded := woc.workflowDeadline != nil && time.Now().UTC().After(*woc.workflowDeadline)
+	if woc.wfSpec.Shutdown != "" || deadlineExceeded {
 		for _, node := range woc.wf.Status.Nodes {
-			if node.IsActiveSuspendNode() {
+			if node.IsActiveSuspendNode() || (node.Phase == wfv1.NodePending && deadlineExceeded) {
 				var message string
 				if woc.wfSpec.Shutdown != "" {
 					message = fmt.Sprintf("Stopped with strategy '%s'", woc.wfSpec.Shutdown)


Test patch:
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 779fda2f0..ea708b22b 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -4434,3 +4434,126 @@ func TestGlobalVarsOnExit(t *testing.T) {
 		assert.Equal(t, "nononono", node.Inputs.Parameters[0].Value.StrVal)
 	}
 }
+
+var deadlineWf = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: steps-9fvnv
+  namespace: argo
+spec:
+  activeDeadlineSeconds: 1
+  entrypoint: main
+  templates:
+  - name: main
+    steps:
+    - - name: approve
+        template: approve
+      - name: hello1
+        template: whalesay
+  - name: approve
+    suspend: {}
+  - container:
+      args:
+      - sleep 50
+      command:
+      - sh
+      - -c
+      image: alpine:latest
+      resources:
+        requests:
+          memory: 1Gi
+    name: whalesay
+    resubmitPendingPods: true
+status:
+  finishedAt: null
+  nodes:
+    steps-9fvnv:
+      children:
+      - steps-9fvnv-3514116232
+      displayName: steps-9fvnv
+      finishedAt: null
+      id: steps-9fvnv
+      name: steps-9fvnv
+      phase: Running
+      startedAt: "2020-07-24T16:39:25Z"
+      templateName: main
+      templateScope: local/steps-9fvnv
+      type: Steps
+    steps-9fvnv-3392004273:
+      boundaryID: steps-9fvnv
+      displayName: hello1
+      finishedAt: null
+      hostNodeName: k3d-k3s-default-server
+      id: steps-9fvnv-3392004273
+      message: 'ErrImageNeverPull: Container image "argoproj/argoexec:latest" is not
+        present with pull policy of Never'
+      name: steps-9fvnv[0].hello1
+      phase: Pending
+      startedAt: "2020-07-24T16:39:25Z"
+      templateName: whalesay
+      templateScope: local/steps-9fvnv
+      type: Pod
+    steps-9fvnv-3514116232:
+      boundaryID: steps-9fvnv
+      children:
+      - steps-9fvnv-3700512507
+      - steps-9fvnv-3392004273
+      displayName: '[0]'
+      finishedAt: null
+      id: steps-9fvnv-3514116232
+      name: steps-9fvnv[0]
+      phase: Running
+      startedAt: "2020-07-24T16:39:25Z"
+      templateName: main
+      templateScope: local/steps-9fvnv
+      type: StepGroup
+    steps-9fvnv-3700512507:
+      boundaryID: steps-9fvnv
+      displayName: approve
+      finishedAt: null
+      id: steps-9fvnv-3700512507
+      name: steps-9fvnv[0].approve
+      phase: Running
+      startedAt: "2020-07-24T16:39:25Z"
+      templateName: approve
+      templateScope: local/steps-9fvnv
+      type: Suspend
+  phase: Running
+  startedAt: "2020-07-24T16:39:25Z"
+`
+
+func TestFailSuspendedAndPendingNodesAfterDeadline(t *testing.T) {
+	wf := unmarshalWF(deadlineWf)
+	wf.Status.StartedAt = metav1.Now()
+	cancel, controller := newController(wf)
+	defer cancel()
+	woc := newWorkflowOperationCtx(wf, controller)
+	t.Run("Before Deadline", func(t *testing.T) {
+		woc.operate()
+		assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+	})
+	time.Sleep(1 * time.Second)
+	t.Run("After Deadline", func(t *testing.T) {
+		woc.operate()
+		assert.Equal(t, wfv1.NodeFailed, woc.wf.Status.Phase)
+		for _, node := range woc.wf.Status.Nodes {
+			assert.Equal(t, wfv1.NodeFailed, node.Phase)
+		}
+	})
+}
+
+func TestFailSuspendedAndPendingNodesAfterShutdown(t *testing.T) {
+	wf := unmarshalWF(deadlineWf)
+	wf.Spec.Shutdown = wfv1.ShutdownStrategyStop
+	cancel, controller := newController(wf)
+	defer cancel()
+	woc := newWorkflowOperationCtx(wf, controller)
+	t.Run("After Shutdown", func(t *testing.T) {
+		woc.operate()
+		assert.Equal(t, wfv1.NodeFailed, woc.wf.Status.Phase)
+		for _, node := range woc.wf.Status.Nodes {
+			assert.Equal(t, wfv1.NodeFailed, node.Phase)
+		}
+	})
+}
\ No newline at end of file