Repository: argoproj/argo (go)

Issue:
CLI/API doesn't properly support workflow compression or node offload
Checklist:

 * [X] I've included the version.
 * [ ] I've included reproduction steps.
 * [ ] I've included the workflow YAML.
 * [ ] I've included the logs.
 
**What happened**:

Argo CLI/API doesn't properly support workflow compression or node offload, for example:
- workflow retry in RetryWorkflow in workflow_server.go doesn't support either of these
- workflow resume supports decompression but not recompression or node offload

**What you expected to happen**:

All API (and therefore CLI) commands to support both

**Environment**:
- Argo version: 2.7

---
<!-- Issue Author: Don't delete this message to ecourage 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: Consider offloaded and compressed node in retry and resume
Fixes #2592

~Still needs tests~ Done

Patch:
diff --git a/persist/sqldb/offload_node_status_repo.go b/persist/sqldb/offload_node_status_repo.go
index e34450517..30b0bcc68 100644
--- a/persist/sqldb/offload_node_status_repo.go
+++ b/persist/sqldb/offload_node_status_repo.go
@@ -15,7 +15,7 @@ import (
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 )
 
-const OffloadNodeStatusDisabledWarning = "Workflow has offloaded nodes, but offloading has been disabled"
+const OffloadNodeStatusDisabled = "Workflow has offloaded nodes, but offloading has been disabled"
 
 type UUIDVersion struct {
 	UID     string `db:"uid"`
diff --git a/server/artifacts/artifact_server.go b/server/artifacts/artifact_server.go
index 8c724d9be..af787fcf7 100644
--- a/server/artifacts/artifact_server.go
+++ b/server/artifacts/artifact_server.go
@@ -176,7 +176,7 @@ func (a *ArtifactServer) getWorkflow(ctx context.Context, namespace string, work
 			}
 			wf.Status.Nodes = offloadedNodes
 		} else {
-			log.WithFields(log.Fields{"namespace": namespace, "name": workflowName}).Warn(sqldb.OffloadNodeStatusDisabledWarning)
+			log.WithFields(log.Fields{"namespace": namespace, "name": workflowName}).Warn(sqldb.OffloadNodeStatusDisabled)
 		}
 	}
 	return wf, nil
diff --git a/server/workflow/workflow_server.go b/server/workflow/workflow_server.go
index 90618cc85..03a72a146 100644
--- a/server/workflow/workflow_server.go
+++ b/server/workflow/workflow_server.go
@@ -102,7 +102,7 @@ func (s *workflowServer) GetWorkflow(ctx context.Context, req *workflowpkg.Workf
 			}
 			wf.Status.Nodes = offloadedNodes
 		} else {
-			log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabledWarning)
+			log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabled)
 		}
 	}
 	err = packer.DecompressWorkflow(wf)
@@ -134,7 +134,7 @@ func (s *workflowServer) ListWorkflows(ctx context.Context, req *workflowpkg.Wor
 				if s.offloadNodeStatusRepo.IsEnabled() {
 					wfList.Items[i].Status.Nodes = offloadedNodes[sqldb.UUIDVersion{UID: string(wf.UID), Version: wf.GetOffloadNodeStatusVersion()}]
 				} else {
-					log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn("Workflow has offloaded nodes, but offloading has been disabled")
+					log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabled)
 				}
 			}
 		}
@@ -182,7 +182,7 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 				}
 				wf.Status.Nodes = offloadedNodes
 			} else {
-				log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabledWarning)
+				log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabled)
 			}
 		}
 		logCtx.Debug("Sending event")
@@ -220,7 +220,7 @@ func (s *workflowServer) RetryWorkflow(ctx context.Context, req *workflowpkg.Wor
 		return nil, err
 	}
 
-	wf, err = util.RetryWorkflow(kubeClient, wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), wf)
+	wf, err = util.RetryWorkflow(kubeClient, s.offloadNodeStatusRepo, wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), wf)
 	if err != nil {
 		return nil, err
 	}
@@ -254,7 +254,7 @@ func (s *workflowServer) ResumeWorkflow(ctx context.Context, req *workflowpkg.Wo
 		return nil, err
 	}
 
-	err = util.ResumeWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), req.Name, req.NodeFieldSelector)
+	err = util.ResumeWorkflow(wfClient.ArgoprojV1alpha1().Workflows(req.Namespace), s.offloadNodeStatusRepo, req.Name, req.NodeFieldSelector)
 	if err != nil {
 		log.Warnf("Failed to resume %s: %+v", req.Name, err)
 		return nil, err
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index af8edec5c..5fe538b4f 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -464,7 +464,7 @@ func (woc *wfOperationCtx) persistUpdates() {
 	// try and compress nodes if needed
 	nodes := woc.wf.Status.Nodes
 
-	err := packer.CompressWorkflow(woc.wf)
+	err := packer.CompressWorkflowIfNeeded(woc.wf)
 	if packer.IsTooLargeError(err) || os.Getenv("ALWAYS_OFFLOAD_NODE_STATUS") == "true" {
 		if woc.controller.offloadNodeStatusRepo.IsEnabled() {
 			offloadVersion, err := woc.controller.offloadNodeStatusRepo.Save(string(woc.wf.UID), woc.wf.Namespace, nodes)
diff --git a/workflow/packer/packer.go b/workflow/packer/packer.go
index 6e16968bf..1de6d6d05 100644
--- a/workflow/packer/packer.go
+++ b/workflow/packer/packer.go
@@ -59,7 +59,7 @@ func IsTooLargeError(err error) bool {
 	return err != nil && strings.HasPrefix(err.Error(), tooLarge)
 }
 
-func CompressWorkflow(wf *wfv1.Workflow) error {
+func CompressWorkflowIfNeeded(wf *wfv1.Workflow) error {
 	large, err := IsLargeWorkflow(wf)
 	if err != nil {
 		return err
@@ -67,6 +67,10 @@ func CompressWorkflow(wf *wfv1.Workflow) error {
 	if !large {
 		return nil
 	}
+	return compressWorkflow(wf)
+}
+
+func compressWorkflow(wf *wfv1.Workflow) error {
 	nodeContent, err := json.Marshal(wf.Status.Nodes)
 	if err != nil {
 		return err
@@ -74,7 +78,7 @@ func CompressWorkflow(wf *wfv1.Workflow) error {
 	wf.Status.CompressedNodes = file.CompressEncodeString(string(nodeContent))
 	wf.Status.Nodes = nil
 	// still too large?
-	large, err = IsLargeWorkflow(wf)
+	large, err := IsLargeWorkflow(wf)
 	if err != nil {
 		return err
 	}
diff --git a/workflow/util/util.go b/workflow/util/util.go
index 3640e5b9f..4528988f9 100644
--- a/workflow/util/util.go
+++ b/workflow/util/util.go
@@ -35,6 +35,7 @@ import (
 	"sigs.k8s.io/yaml"
 
 	"github.com/argoproj/argo/errors"
+	"github.com/argoproj/argo/persist/sqldb"
 	"github.com/argoproj/argo/pkg/apis/workflow"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	wfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned"
@@ -341,7 +342,7 @@ func SuspendWorkflow(wfIf v1alpha1.WorkflowInterface, workflowName string) error
 
 // ResumeWorkflow resumes a workflow by setting spec.suspend to nil and any suspended nodes to Successful.
 // Retries conflict errors
-func ResumeWorkflow(wfIf v1alpha1.WorkflowInterface, workflowName string, nodeFieldSelector string) error {
+func ResumeWorkflow(wfIf v1alpha1.WorkflowInterface, repo sqldb.OffloadNodeStatusRepo, workflowName string, nodeFieldSelector string) error {
 	if len(nodeFieldSelector) > 0 {
 		return updateWorkflowNodeByKey(wfIf, workflowName, nodeFieldSelector, wfv1.NodeSucceeded, "")
 	} else {
@@ -353,7 +354,7 @@ func ResumeWorkflow(wfIf v1alpha1.WorkflowInterface, workflowName string, nodeFi
 
 			err = packer.DecompressWorkflow(wf)
 			if err != nil {
-				log.Fatal(err)
+				return false, fmt.Errorf("unable to decompress workflow: %s", err)
 			}
 
 			workflowUpdated := false
@@ -362,17 +363,50 @@ func ResumeWorkflow(wfIf v1alpha1.WorkflowInterface, workflowName string, nodeFi
 				workflowUpdated = true
 			}
 
+			nodes := wf.Status.Nodes
+			if wf.Status.IsOffloadNodeStatus() {
+				if !repo.IsEnabled() {
+					return false, fmt.Errorf(sqldb.OffloadNodeStatusDisabled)
+				}
+				var err error
+				nodes, err = repo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
+				if err != nil {
+					return false, fmt.Errorf("unable to retrieve offloaded nodes: %s", err)
+				}
+			}
+			newNodes := nodes.DeepCopy()
+
 			// To resume a workflow with a suspended node we simply mark the node as Successful
-			for nodeID, node := range wf.Status.Nodes {
+			for nodeID, node := range nodes {
 				if node.IsActiveSuspendNode() {
 					node.Phase = wfv1.NodeSucceeded
 					node.FinishedAt = metav1.Time{Time: time.Now().UTC()}
-					wf.Status.Nodes[nodeID] = node
+					newNodes[nodeID] = node
 					workflowUpdated = true
 				}
 			}
 
 			if workflowUpdated {
+				if wf.Status.IsOffloadNodeStatus() {
+					if !repo.IsEnabled() {
+						return false, fmt.Errorf(sqldb.OffloadNodeStatusDisabled)
+					}
+					offloadVersion, err := repo.Save(string(wf.UID), wf.Namespace, newNodes)
+					if err != nil {
+						return false, fmt.Errorf("unable to save offloaded nodes: %s", err)
+					}
+					wf.Status.OffloadNodeStatusVersion = offloadVersion
+					wf.Status.CompressedNodes = ""
+					wf.Status.Nodes = nil
+				} else {
+					wf.Status.Nodes = newNodes
+				}
+
+				err = packer.CompressWorkflowIfNeeded(wf)
+				if err != nil {
+					return false, fmt.Errorf("unable to compress workflow: %s", err)
+				}
+
 				_, err = wfIf.Update(wf)
 				if err != nil {
 					if apierr.IsConflict(err) {
@@ -578,12 +612,18 @@ func convertNodeID(newWf *wfv1.Workflow, regex *regexp.Regexp, oldNodeID string,
 }
 
 // RetryWorkflow updates a workflow, deleting all failed steps as well as the onExit node (and children)
-func RetryWorkflow(kubeClient kubernetes.Interface, wfClient v1alpha1.WorkflowInterface, wf *wfv1.Workflow) (*wfv1.Workflow, error) {
+func RetryWorkflow(kubeClient kubernetes.Interface, repo sqldb.OffloadNodeStatusRepo, wfClient v1alpha1.WorkflowInterface, wf *wfv1.Workflow) (*wfv1.Workflow, error) {
 	switch wf.Status.Phase {
 	case wfv1.NodeFailed, wfv1.NodeError:
 	default:
 		return nil, errors.Errorf(errors.CodeBadRequest, "workflow must be Failed/Error to retry")
 	}
+
+	err := packer.DecompressWorkflow(wf)
+	if err != nil {
+		return nil, fmt.Errorf("unable to decompress workflow: %s", err)
+	}
+
 	newWF := wf.DeepCopy()
 	podIf := kubeClient.CoreV1().Pods(wf.ObjectMeta.Namespace)
 
@@ -600,22 +640,34 @@ func RetryWorkflow(kubeClient kubernetes.Interface, wfClient v1alpha1.WorkflowIn
 	}
 
 	// Iterate the previous nodes. If it was successful Pod carry it forward
-	newWF.Status.Nodes = make(map[string]wfv1.NodeStatus)
+	newNodes := make(map[string]wfv1.NodeStatus)
 	onExitNodeName := wf.ObjectMeta.Name + ".onExit"
-	for _, node := range wf.Status.Nodes {
+	nodes := wf.Status.Nodes
+	if wf.Status.IsOffloadNodeStatus() {
+		if !repo.IsEnabled() {
+			return nil, fmt.Errorf(sqldb.OffloadNodeStatusDisabled)
+		}
+		var err error
+		nodes, err = repo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
+		if err != nil {
+			return nil, fmt.Errorf("unable to retrieve offloaded nodes: %s", err)
+		}
+	}
+
+	for _, node := range nodes {
 		switch node.Phase {
 		case wfv1.NodeSucceeded, wfv1.NodeSkipped:
 			if !strings.HasPrefix(node.Name, onExitNodeName) {
-				newWF.Status.Nodes[node.ID] = node
+				newNodes[node.ID] = node
 				continue
 			}
 		case wfv1.NodeError, wfv1.NodeFailed:
-			if !strings.HasPrefix(node.Name, onExitNodeName) && node.Type == wfv1.NodeTypeDAG {
+			if !strings.HasPrefix(node.Name, onExitNodeName) && (node.Type == wfv1.NodeTypeDAG || node.Type == wfv1.NodeTypeStepGroup) {
 				newNode := node.DeepCopy()
 				newNode.Phase = wfv1.NodeRunning
 				newNode.Message = ""
 				newNode.FinishedAt = metav1.Time{}
-				newWF.Status.Nodes[newNode.ID] = *newNode
+				newNodes[newNode.ID] = *newNode
 				continue
 			}
 			// do not add this status to the node. pretend as if this node never existed.
@@ -634,16 +686,36 @@ func RetryWorkflow(kubeClient kubernetes.Interface, wfClient v1alpha1.WorkflowIn
 			newNode.Phase = wfv1.NodeRunning
 			newNode.Message = ""
 			newNode.FinishedAt = metav1.Time{}
-			newWF.Status.Nodes[newNode.ID] = *newNode
+			newNodes[newNode.ID] = *newNode
 			continue
 		}
 	}
 
+	if wf.Status.IsOffloadNodeStatus() {
+		if !repo.IsEnabled() {
+			return nil, fmt.Errorf(sqldb.OffloadNodeStatusDisabled)
+		}
+		offloadVersion, err := repo.Save(string(newWF.UID), newWF.Namespace, newNodes)
+		if err != nil {
+			return nil, fmt.Errorf("unable to save offloaded nodes: %s", err)
+		}
+		newWF.Status.OffloadNodeStatusVersion = offloadVersion
+		newWF.Status.CompressedNodes = ""
+		newWF.Status.Nodes = nil
+	} else {
+		newWF.Status.Nodes = newNodes
+	}
+
 	newWF.Status.StoredTemplates = make(map[string]wfv1.Template)
 	for id, tmpl := range wf.Status.StoredTemplates {
 		newWF.Status.StoredTemplates[id] = tmpl
 	}
 
+	err = packer.CompressWorkflowIfNeeded(newWF)
+	if err != nil {
+		return nil, fmt.Errorf("unable to compress workflow: %s", err)
+	}
+
 	return wfClient.Update(newWF)
 }
 


Test patch:
diff --git a/test/e2e/cli_test.go b/test/e2e/cli_test.go
index db7a654e1..0bc14d30f 100644
--- a/test/e2e/cli_test.go
+++ b/test/e2e/cli_test.go
@@ -470,7 +470,12 @@ func (s *CLISuite) TestWorkflowLint() {
 	})
 }
 
-func (s *CLISuite) TestWorkflowRetry() {
+func (s *CLISuite) TestWorkflowRetryNoPersistence() {
+	if s.Persistence.IsEnabled() {
+		// Persistence is enabled for this test, but it is not enabled for the Argo Server in this test suite.
+		// When this is the case, this behavior is tested in cli_with_server_test.go
+		s.T().SkipNow()
+	}
 	s.Given().
 		Workflow("@testdata/exit-1.yaml").
 		When().
diff --git a/test/e2e/cli_with_server_test.go b/test/e2e/cli_with_server_test.go
index 248d960f5..b8286b6d4 100644
--- a/test/e2e/cli_with_server_test.go
+++ b/test/e2e/cli_with_server_test.go
@@ -97,6 +97,26 @@ func (s *CLIWithServerSuite) TestArchive() {
 	})
 }
 
+func (s *CLIWithServerSuite) TestWorkflowRetryPersistence() {
+	if !s.Persistence.IsEnabled() {
+		// Persistence is disabled for this test, but it is enabled for the Argo Server in this test suite.
+		// When this is the case, this behavior is tested in cli_test.go
+		s.T().SkipNow()
+	}
+	s.Given().
+		Workflow("@testdata/exit-1.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(20*time.Second).
+		Given().
+		RunCli([]string{"retry", "exit-1"}, func(t *testing.T, output string, err error) {
+			if assert.NoError(t, err) {
+				assert.Contains(t, output, "Name:")
+				assert.Contains(t, output, "Namespace:")
+			}
+		})
+}
+
 func TestCLIWithServerSuite(t *testing.T) {
 	suite.Run(t, new(CLIWithServerSuite))
 }
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index e34c6edaf..5059078bd 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -14,6 +14,7 @@ import (
 	"sigs.k8s.io/yaml"
 
 	"github.com/argoproj/argo/config"
+	"github.com/argoproj/argo/persist/sqldb"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/test"
 	"github.com/argoproj/argo/workflow/common"
@@ -847,7 +848,7 @@ func TestSuspendResume(t *testing.T) {
 	assert.Equal(t, 0, len(pods.Items))
 
 	// resume the workflow and operate again. two pods should be able to be scheduled
-	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "")
+	err = util.ResumeWorkflow(wfcset, sqldb.ExplosiveOffloadNodeStatusRepo, wf.ObjectMeta.Name, "")
 	assert.NoError(t, err)
 	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
 	assert.NoError(t, err)
@@ -1166,7 +1167,7 @@ func TestSuspendTemplate(t *testing.T) {
 	assert.Equal(t, 0, len(pods.Items))
 
 	// resume the workflow. verify resume workflow edits nodestatus correctly
-	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "")
+	err = util.ResumeWorkflow(wfcset, sqldb.ExplosiveOffloadNodeStatusRepo, wf.ObjectMeta.Name, "")
 	assert.NoError(t, err)
 	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
 	assert.NoError(t, err)
@@ -1239,7 +1240,7 @@ func TestSuspendTemplateWithFilteredResume(t *testing.T) {
 	assert.Equal(t, 0, len(pods.Items))
 
 	// resume the workflow, but with non-matching selector
-	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "inputs.paramaters.param1.value=value2")
+	err = util.ResumeWorkflow(wfcset, sqldb.ExplosiveOffloadNodeStatusRepo, wf.ObjectMeta.Name, "inputs.paramaters.param1.value=value2")
 	assert.Error(t, err)
 
 	// operate the workflow. nothing should have happened
@@ -1251,7 +1252,7 @@ func TestSuspendTemplateWithFilteredResume(t *testing.T) {
 	assert.True(t, util.IsWorkflowSuspended(wf))
 
 	// resume the workflow, but with matching selector
-	err = util.ResumeWorkflow(wfcset, wf.ObjectMeta.Name, "inputs.parameters.param1.value=value1")
+	err = util.ResumeWorkflow(wfcset, sqldb.ExplosiveOffloadNodeStatusRepo, wf.ObjectMeta.Name, "inputs.parameters.param1.value=value1")
 	assert.NoError(t, err)
 	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
 	assert.NoError(t, err)
diff --git a/workflow/packer/packer_test.go b/workflow/packer/packer_test.go
index b7217c9bf..27b3aaca2 100644
--- a/workflow/packer/packer_test.go
+++ b/workflow/packer/packer_test.go
@@ -21,7 +21,7 @@ func TestDecompressWorkflow(t *testing.T) {
 				Nodes: wfv1.Nodes{"foo": wfv1.NodeStatus{}},
 			},
 		}
-		err := CompressWorkflow(wf)
+		err := CompressWorkflowIfNeeded(wf)
 		if assert.NoError(t, err) {
 			assert.NotNil(t, wf)
 			assert.NotEmpty(t, wf.Status.Nodes)
@@ -40,7 +40,7 @@ func TestDecompressWorkflow(t *testing.T) {
 				Nodes: wfv1.Nodes{"foo": wfv1.NodeStatus{}, "bar": wfv1.NodeStatus{}},
 			},
 		}
-		err := CompressWorkflow(wf)
+		err := CompressWorkflowIfNeeded(wf)
 		if assert.NoError(t, err) {
 			assert.NotNil(t, wf)
 			assert.Empty(t, wf.Status.Nodes)
@@ -62,7 +62,7 @@ func TestDecompressWorkflow(t *testing.T) {
 				Nodes: wfv1.Nodes{"foo": wfv1.NodeStatus{}, "bar": wfv1.NodeStatus{}, "baz": wfv1.NodeStatus{}, "qux": wfv1.NodeStatus{}},
 			},
 		}
-		err := CompressWorkflow(wf)
+		err := CompressWorkflowIfNeeded(wf)
 		if assert.Error(t, err) {
 			assert.True(t, IsTooLargeError(err))
 			// if too large, we want the original back please
diff --git a/workflow/util/util_test.go b/workflow/util/util_test.go
index bf5916b92..0f28c6ab0 100644
--- a/workflow/util/util_test.go
+++ b/workflow/util/util_test.go
@@ -9,11 +9,16 @@ import (
 
 	"github.com/ghodss/yaml"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
 
+	"github.com/argoproj/argo/persist/sqldb"
+	"github.com/argoproj/argo/persist/sqldb/mocks"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	fakeClientset "github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
+	"github.com/argoproj/argo/workflow/packer"
 )
 
 // TestSubmitDryRun
@@ -199,3 +204,287 @@ func TestPodSpecPatchMerge(t *testing.T) {
 	assert.Equal(t, "1.000", spec.Containers[0].Resources.Limits.Cpu().AsDec().String())
 	assert.Equal(t, "104857600", spec.Containers[0].Resources.Limits.Memory().AsDec().String())
 }
+
+var suspendedWf = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  creationTimestamp: "2020-04-10T15:21:23Z"
+  name: suspend
+  generation: 2
+  labels:
+    workflows.argoproj.io/phase: Running
+  resourceVersion: "238969"
+  selfLink: /apis/argoproj.io/v1alpha1/namespaces/argo/workflows/suspend
+  uid: 4f08d325-dc5a-43a3-9986-259e259e6ea3
+spec:
+  arguments: {}
+  entrypoint: suspend
+  templates:
+  - arguments: {}
+    inputs: {}
+    metadata: {}
+    name: suspend
+    outputs: {}
+    steps:
+    - - arguments: {}
+        name: approve
+        template: approve
+  - arguments: {}
+    inputs: {}
+    metadata: {}
+    name: approve
+    outputs: {}
+    suspend: {}
+status:
+  finishedAt: null
+  nodes:
+    suspend-template-xjsg2:
+      children:
+      - suspend-template-xjsg2-4125372399
+      displayName: suspend-template-xjsg2
+      finishedAt: null
+      id: suspend-template-xjsg2
+      name: suspend-template-xjsg2
+      phase: Running
+      startedAt: "2020-04-10T15:21:23Z"
+      templateName: suspend
+      templateScope: local/suspend-template-xjsg2
+      type: Steps
+    suspend-template-xjsg2-1771269240:
+      boundaryID: suspend-template-xjsg2
+      displayName: approve
+      finishedAt: null
+      id: suspend-template-xjsg2-1771269240
+      name: suspend-template-xjsg2[0].approve
+      phase: Running
+      startedAt: "2020-04-10T15:21:23Z"
+      templateName: approve
+      templateScope: local/suspend-template-xjsg2
+      type: Suspend
+    suspend-template-xjsg2-4125372399:
+      boundaryID: suspend-template-xjsg2
+      children:
+      - suspend-template-xjsg2-1771269240
+      displayName: '[0]'
+      finishedAt: null
+      id: suspend-template-xjsg2-4125372399
+      name: suspend-template-xjsg2[0]
+      phase: Running
+      startedAt: "2020-04-10T15:21:23Z"
+      templateName: suspend
+      templateScope: local/suspend-template-xjsg2
+      type: StepGroup
+  phase: Running
+  startedAt: "2020-04-10T15:21:23Z"
+`
+
+func TestResumeWorkflowCompressed(t *testing.T) {
+	wfIf := fakeClientset.NewSimpleClientset().ArgoprojV1alpha1().Workflows("")
+	origWf := unmarshalWF(suspendedWf)
+
+	clearFunc := packer.SetMaxWorkflowSize(1156)
+	defer clearFunc()
+	err := packer.CompressWorkflowIfNeeded(origWf)
+	assert.NoError(t, err)
+
+	_, err = wfIf.Create(origWf)
+	assert.NoError(t, err)
+
+	err = ResumeWorkflow(wfIf, sqldb.ExplosiveOffloadNodeStatusRepo, "suspend", "")
+	assert.NoError(t, err)
+
+	wf, err := wfIf.Get("suspend", metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.NotEmpty(t, wf.Status.CompressedNodes)
+}
+
+func TestResumeWorkflowOffloaded(t *testing.T) {
+	wfIf := fakeClientset.NewSimpleClientset().ArgoprojV1alpha1().Workflows("")
+	origWf := unmarshalWF(suspendedWf)
+
+	origNodes := origWf.Status.Nodes
+
+	origWf.Status.Nodes = nil
+	origWf.Status.OffloadNodeStatusVersion = "123"
+
+	_, err := wfIf.Create(origWf)
+	assert.NoError(t, err)
+
+	offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{}
+	offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true)
+	offloadNodeStatusRepo.On("Get", "4f08d325-dc5a-43a3-9986-259e259e6ea3", "123").Return(origNodes, nil)
+	offloadNodeStatusRepo.On("Save", "4f08d325-dc5a-43a3-9986-259e259e6ea3", mock.Anything, mock.Anything).Return("1234", nil)
+
+	err = ResumeWorkflow(wfIf, offloadNodeStatusRepo, "suspend", "")
+	assert.NoError(t, err)
+
+	wf, err := wfIf.Get("suspend", metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, "1234", wf.Status.OffloadNodeStatusVersion)
+}
+
+var failedWf = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  creationTimestamp: "2020-04-10T15:35:59Z"
+  generation: 5
+  labels:
+    workflows.argoproj.io/completed: "true"
+    workflows.argoproj.io/phase: Failed
+  name: fail-template
+  resourceVersion: "240144"
+  selfLink: /apis/argoproj.io/v1alpha1/namespaces/argo/workflows/fail-template
+  uid: 7e74dbb9-d681-4c22-9bed-a581ec28383f
+spec:
+  arguments: {}
+  entrypoint: fail
+  templates:
+  - arguments: {}
+    inputs: {}
+    metadata: {}
+    name: fail
+    outputs: {}
+    steps:
+    - - arguments: {}
+        name: approve
+        template: approve
+  - arguments: {}
+    container:
+      args:
+      - exit 1
+      command:
+      - sh
+      - -c
+      image: alpine:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: approve
+    outputs: {}
+status:
+  conditions:
+  - status: "True"
+    type: Completed
+  finishedAt: "2020-04-10T15:36:03Z"
+  message: child 'fail-template-2878444447' failed
+  nodes:
+    fail-template:
+      children:
+      - fail-template-2384955452
+      displayName: fail-template
+      finishedAt: "2020-04-10T15:36:03Z"
+      id: fail-template
+      message: child 'fail-template-2878444447' failed
+      name: fail-template
+      outboundNodes:
+      - fail-template-2878444447
+      phase: Failed
+      startedAt: "2020-04-10T15:35:59Z"
+      templateName: fail
+      templateScope: local/fail-template
+      type: Steps
+    fail-template-2384955452:
+      boundaryID: fail-template
+      children:
+      - fail-template-2878444447
+      displayName: '[0]'
+      finishedAt: "2020-04-10T15:36:03Z"
+      id: fail-template-2384955452
+      message: child 'fail-template-2878444447' failed
+      name: fail-template[0]
+      phase: Failed
+      startedAt: "2020-04-10T15:35:59Z"
+      templateName: fail
+      templateScope: local/fail-template
+      type: StepGroup
+    fail-template-2878444447:
+      boundaryID: fail-template
+      displayName: approve
+      finishedAt: "2020-04-10T15:36:02Z"
+      id: fail-template-2878444447
+      message: failed with exit code 1
+      name: fail-template[0].approve
+      outputs:
+        artifacts:
+        - archiveLogs: true
+          name: main-logs
+          s3:
+            accessKeySecret:
+              key: accesskey
+              name: my-minio-cred
+            bucket: my-bucket
+            endpoint: minio:9000
+            insecure: true
+            key: fail-template/fail-template-2878444447/main.log
+            secretKeySecret:
+              key: secretkey
+              name: my-minio-cred
+      phase: Failed
+      resourcesDuration:
+        cpu: 2
+        memory: 0
+      startedAt: "2020-04-10T15:35:59Z"
+      templateName: approve
+      templateScope: local/fail-template
+      type: Pod
+  phase: Failed
+  resourcesDuration:
+    cpu: 2
+    memory: 0
+  startedAt: "2020-04-10T15:35:59Z"
+
+`
+
+func TestRetryWorkflowCompressed(t *testing.T) {
+	wfIf := fakeClientset.NewSimpleClientset().ArgoprojV1alpha1().Workflows("")
+	origWf := unmarshalWF(failedWf)
+	kubeCs := fake.NewSimpleClientset()
+
+	_, err := kubeCs.CoreV1().Pods("").Create(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "fail-template-2878444447"}})
+	assert.NoError(t, err)
+
+	clearFunc := packer.SetMaxWorkflowSize(1684)
+	defer clearFunc()
+	err = packer.CompressWorkflowIfNeeded(origWf)
+	assert.NoError(t, err)
+
+	_, err = wfIf.Create(origWf)
+	assert.NoError(t, err)
+
+	clearFunc = packer.SetMaxWorkflowSize(1557)
+	defer clearFunc()
+	wf, err := RetryWorkflow(kubeCs, sqldb.ExplosiveOffloadNodeStatusRepo, wfIf, origWf)
+	assert.NoError(t, err)
+	assert.NotEmpty(t, wf.Status.CompressedNodes)
+}
+
+func TestRetryWorkflowOffloaded(t *testing.T) {
+	wfIf := fakeClientset.NewSimpleClientset().ArgoprojV1alpha1().Workflows("")
+	origWf := unmarshalWF(failedWf)
+	kubeCs := fake.NewSimpleClientset()
+	_, err := kubeCs.CoreV1().Pods("").Create(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "fail-template-2878444447"}})
+	assert.NoError(t, err)
+
+	origNodes := origWf.Status.Nodes
+
+	origWf.Status.Nodes = nil
+	origWf.Status.OffloadNodeStatusVersion = "123"
+
+	_, err = wfIf.Create(origWf)
+	assert.NoError(t, err)
+
+	offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{}
+	offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true)
+	offloadNodeStatusRepo.On("Get", "7e74dbb9-d681-4c22-9bed-a581ec28383f", "123").Return(origNodes, nil)
+	offloadNodeStatusRepo.On("Save", "7e74dbb9-d681-4c22-9bed-a581ec28383f", mock.Anything, mock.Anything).Return("1234", nil)
+
+	_, err = RetryWorkflow(kubeCs, offloadNodeStatusRepo, wfIf, origWf)
+	assert.NoError(t, err)
+
+	wf, err := wfIf.Get("fail-template", metav1.GetOptions{})
+	assert.NoError(t, err)
+	assert.Equal(t, "1234", wf.Status.OffloadNodeStatusVersion)
+}