From ec856835a3a4ec78164aa737f98d4b1653809781 Mon Sep 17 00:00:00 2001
From: Yuan Tang <terrytangyuan@gmail.com>
Date: Fri, 17 Mar 2023 12:33:20 -0400
Subject: [PATCH] fix: PVC in wf.status should be reset when retrying workflow
 (#10685)

Signed-off-by: Yuan Tang <terrytangyuan@gmail.com>
---
 test/e2e/cli_test.go                          | 24 +++++++++++
 .../retry-with-recreated-pvc-test.yaml        | 42 +++++++++++++++++++
 workflow/util/util.go                         |  1 +
 3 files changed, 67 insertions(+)
 create mode 100644 test/e2e/testdata/retry-with-recreated-pvc-test.yaml

diff --git a/test/e2e/cli_test.go b/test/e2e/cli_test.go
index cb4449642f03..6ad02f175306 100644
--- a/test/e2e/cli_test.go
+++ b/test/e2e/cli_test.go
@@ -890,6 +890,30 @@ func (s *CLISuite) TestWorkflowRetryNestedDag() {
 		})
 }
 
+func (s *CLISuite) TestWorkflowRetryWithRecreatedPVC() {
+	s.Given().
+		Workflow("@testdata/retry-with-recreated-pvc-test.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(fixtures.ToBeFailed).
+		Then().
+		RunCli([]string{"retry", "retry-with-recreated-pvc"}, func(t *testing.T, output string, err error) {
+			if assert.NoError(t, err, output) {
+				assert.Contains(t, output, "Name:")
+				assert.Contains(t, output, "Namespace:")
+			}
+		}).
+		When().
+		WaitForWorkflow(fixtures.ToBeFailed).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeFailed, status.Nodes.FindByDisplayName("print").Phase)
+			// This step is failed intentionally to allow retry. The error message is not related to PVC that is deleted
+			// previously since it is re-created during retry.
+			assert.Equal(t, "Error (exit code 1)", status.Nodes.FindByDisplayName("print").Message)
+		})
+}
+
 func (s *CLISuite) TestWorkflowStop() {
 	s.Given().
 		Workflow("@smoke/basic.yaml").
diff --git a/test/e2e/testdata/retry-with-recreated-pvc-test.yaml b/test/e2e/testdata/retry-with-recreated-pvc-test.yaml
new file mode 100644
index 000000000000..05efa3fe37b5
--- /dev/null
+++ b/test/e2e/testdata/retry-with-recreated-pvc-test.yaml
@@ -0,0 +1,42 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: retry-with-recreated-pvc
+spec:
+  volumeClaimGC:
+    strategy: OnWorkflowCompletion
+  entrypoint: volumes-pvc-example
+  volumeClaimTemplates:
+    - metadata:
+        name: workdir
+      spec:
+        accessModes: [ "ReadWriteOnce" ]
+        resources:
+          requests:
+            storage: 1Gi
+
+  templates:
+    - name: volumes-pvc-example
+      steps:
+        - - name: generate
+            template: whalesay
+        - - name: print
+            template: print-message
+
+    - name: whalesay
+      container:
+        image: argoproj/argosay:v2
+        command: [sh, -c]
+        args: ["echo generating message in volume; cowsay hello world | tee /mnt/vol/hello_world.txt"]
+        volumeMounts:
+          - name: workdir
+            mountPath: /mnt/vol
+
+    - name: print-message
+      container:
+        image: argoproj/argosay:v2
+        command: [sh, -c]
+        args: ["exit 1"]
+        volumeMounts:
+          - name: workdir
+            mountPath: /mnt/vol
diff --git a/workflow/util/util.go b/workflow/util/util.go
index eef49def4fc7..e5605674f754 100644
--- a/workflow/util/util.go
+++ b/workflow/util/util.go
@@ -821,6 +821,7 @@ func FormulateRetryWorkflow(ctx context.Context, wf *wfv1.Workflow, restartSucce
 	newWF.Status.StartedAt = metav1.Time{Time: time.Now().UTC()}
 	newWF.Status.FinishedAt = metav1.Time{}
 	newWF.Spec.Shutdown = ""
+	newWF.Status.PersistentVolumeClaims = []apiv1.Volume{}
 	if newWF.Spec.ActiveDeadlineSeconds != nil && *newWF.Spec.ActiveDeadlineSeconds == 0 {
 		// if it was terminated, unset the deadline
 		newWF.Spec.ActiveDeadlineSeconds = nil