Repository: argoproj/argo (go)

Issue:
Re-submitted workflows do not carry over labels
When you re-submit a workflow, labels for templates should remain?

workflows.argoproj.io/resubmitted-from-workflow

Pull request:
fix(controller): Carry-over labels for re-submitted workflows. Fixes #3622
Checklist:

* [x] 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.
* [x] 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"`.  
* [x] I've signed the CLA.
* [x] 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. 
* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).

Fixes #3622

Patch:
diff --git a/Makefile b/Makefile
index b13009947..6f5f9f410 100644
--- a/Makefile
+++ b/Makefile
@@ -46,7 +46,7 @@ STATIC_BUILD          ?= true
 CI                    ?= false
 PROFILE               ?= minimal
 # whether or not to start the Argo Service in TLS mode
-SECURE                := true
+SECURE                := false
 AUTH_MODE             := hybrid
 ifeq ($(PROFILE),sso)
 AUTH_MODE             := sso
diff --git a/docs/configure-artifact-repository.md b/docs/configure-artifact-repository.md
index 08d32561f..47c8ee298 100644
--- a/docs/configure-artifact-repository.md
+++ b/docs/configure-artifact-repository.md
@@ -142,7 +142,7 @@ access is on a per project rather than per bucket basis.
   (https://console.cloud.google.com/storage/settings).
 - Enable interoperability access if needed.
 - Create a new key if needed.
-- Confiture `s3` artifact as following exmaple.
+- Configure `s3` artifact as following exmaple.
 
 ```yaml
 artifacts:
@@ -164,19 +164,51 @@ artifacts:
         key: secretKey
 ```
 
+## Configuring Alibaba Cloud OSS (Object Storage Service)
+
+To configure artifact storage for Alibaba Cloud OSS, please first follow
+the [official documentation](https://www.alibabacloud.com/product/oss) to set up
+an OSS account and bucket. 
+
+Once it's set up, you can find endpoint and bucket
+information on your OSS dashboard and then use them like the following to
+configure the artifact storage for your workflow:
+
+```yaml
+artifacts:
+  - name: my-art
+    path: /my-artifact
+    oss:
+      endpoint: http://oss-cn-hangzhou-zmf.aliyuncs.com
+      bucket: test-bucket-name
+      key: test/mydirectory/ # this is path in the bucket
+      # accessKeySecret and secretKeySecret are secret selectors.
+      # It references the k8s secret named 'my-oss-credentials'.
+      # This secret is expected to have have the keys 'accessKey'
+      # and 'secretKey', containing the base64 encoded credentials
+      # to the bucket.
+      accessKeySecret:
+        name: my-oss-credentials
+        key: accessKey
+      secretKeySecret:
+        name: my-oss-credentials
+        key: secretKey
+```
+
 # Configure the Default Artifact Repository
 
 In order for Argo to use your artifact repository, you can configure it as the
 default repository. Edit the workflow-controller config map with the correct
 endpoint and access/secret keys for your repository.
 
-## S3 compatible artifact repository bucket (such as AWS, GCS and Minio)
+## S3 compatible artifact repository bucket (such as AWS, GCS, Minio, and Alibaba Cloud OSS)
 
-Use the `endpoint` corresponding to your S3 provider:
+Use the `endpoint` corresponding to your provider:
 
 - AWS: s3.amazonaws.com
 - GCS: storage.googleapis.com
 - Minio: my-minio-endpoint.default:9000
+- Alibaba Cloud OSS: oss-cn-hangzhou-zmf.aliyuncs.com
 
 The `key` is name of the object in the `bucket` The `accessKeySecret` and
 `secretKeySecret` are secret selectors that reference the specified kubernetes
@@ -184,7 +216,7 @@ secret. The secret is expected to have the keys 'accessKey' and 'secretKey',
 containing the base64 encoded credentials to the bucket.
 
 For AWS, the `accessKeySecret` and `secretKeySecret` correspond to
-AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY respectively.
+`AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` respectively.
 
 EC2 provides a metadata API via which applications using the AWS SDK may assume
 IAM roles associated with the instance. If you are running argo on EC2 and the
@@ -204,6 +236,9 @@ project rather than per bucket basis.
 For Minio, the `accessKeySecret` and `secretKeySecret` naturally correspond the
 AccessKey and SecretKey.
 
+For Alibaba Cloud OSS, the `accessKeySecret` and `secretKeySecret` corresponds to
+`accessKeyID` `and accessKeySecret` respectively.
+
 Example:
 
 ```
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index 8e4ff1b31..aa1e844df 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -213,6 +213,8 @@ data:
         parallelism: 3
 
     # SSO Configuration for the Argo server.
+    # You must also start argo server with `--auth-mode sso`.
+    # https://github.com/argoproj/argo/blob/master/docs/argo-server-auth-mode.md
     sso:
       # This is the root URL of the OIDC provider (required).
       issuer: https://issuer.root.url/
diff --git a/server/workflow/workflow_server.go b/server/workflow/workflow_server.go
index 2aaa20c0c..5df28cac3 100644
--- a/server/workflow/workflow_server.go
+++ b/server/workflow/workflow_server.go
@@ -142,11 +142,10 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 	opts := &metav1.ListOptions{}
 	if req.ListOptions != nil {
 		opts = req.ListOptions
-		wfName, err := argoutil.RecoverWorkflowNameFromSelectorString(opts.FieldSelector)
-		if err != nil {
-			return fmt.Errorf("malformed request: workflows must be specified with a 'metadata.name=...' field selector; unable to parse: %w", err)
-		}
-		if len(wfName) > 0 {
+		wfName := argoutil.RecoverWorkflowNameFromSelectorStringIfAny(opts.FieldSelector)
+		if wfName != "" {
+			// If we are using an alias (such as `@latest`) we need to dereference it.
+			// s.getWorkflow does that for us
 			wf, err := s.getWorkflow(wfClient, req.Namespace, wfName, metav1.GetOptions{})
 			if err != nil {
 				return err
diff --git a/util/util.go b/util/util.go
index dc18f8bde..8ab63c6c6 100644
--- a/util/util.go
+++ b/util/util.go
@@ -96,22 +96,21 @@ func RecoverIndexFromNodeName(name string) int {
 
 func GenerateFieldSelectorFromWorkflowName(wfName string) string {
 	result := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", wfName)).String()
-	compare, err := RecoverWorkflowNameFromSelectorString(result)
-	if err != nil || wfName != compare {
-		log.WithFields(log.Fields{"wfName": wfName}).Error(err)
+	compare := RecoverWorkflowNameFromSelectorStringIfAny(result)
+	if wfName != compare {
 		panic(fmt.Sprintf("Could not recover field selector from workflow name. Expected '%s' but got '%s'\n", wfName, compare))
 	}
 	return result
 }
 
-func RecoverWorkflowNameFromSelectorString(selector string) (string, error) {
-	nameIndex := strings.Index(selector, "=")
-	if nameIndex < 0 {
-		return "", fmt.Errorf("incorrect prefix: selector is not formatted correctly")
-	}
-	prefix := selector[:nameIndex]
-	if prefix != "metadata.name" {
-		return "", fmt.Errorf("incorrect prefix: expected 'metadata.name' but got '%s'", prefix)
+func RecoverWorkflowNameFromSelectorStringIfAny(selector string) string {
+	const tag = "metadata.name="
+	if starts := strings.Index(selector, tag); starts > -1 {
+		suffix := selector[starts+len(tag):]
+		if ends := strings.Index(suffix, ","); ends > -1 {
+			return strings.TrimSpace(suffix[:ends])
+		}
+		return strings.TrimSpace(suffix)
 	}
-	return selector[nameIndex+1:], nil
+	return ""
 }
diff --git a/workflow/common/util.go b/workflow/common/util.go
index 44089adb0..c719ff8b2 100644
--- a/workflow/common/util.go
+++ b/workflow/common/util.go
@@ -307,7 +307,10 @@ func SubstituteParams(tmpl *wfv1.Template, globalParams, localParams Parameters)
 	}
 	// First replace globals & locals, then replace inputs because globals could be referenced in the inputs
 	replaceMap := globalParams.Merge(localParams)
-	fstTmpl := fasttemplate.New(string(tmplBytes), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(tmplBytes), "{{", "}}")
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	globalReplacedTmplStr, err := Replace(fstTmpl, replaceMap, true)
 	if err != nil {
 		return nil, err
@@ -347,7 +350,10 @@ func SubstituteParams(tmpl *wfv1.Template, globalParams, localParams Parameters)
 		}
 	}
 
-	fstTmpl = fasttemplate.New(globalReplacedTmplStr, "{{", "}}")
+	fstTmpl, err = fasttemplate.NewTemplate(globalReplacedTmplStr, "{{", "}}")
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	s, err := Replace(fstTmpl, replaceMap, true)
 	if err != nil {
 		return nil, err
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index baced0118..447620d69 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -502,7 +502,10 @@ func (woc *wfOperationCtx) resolveDependencyReferences(dagCtx *dagContext, task
 	if err != nil {
 		return nil, errors.InternalWrapError(err)
 	}
-	fstTmpl := fasttemplate.New(string(taskBytes), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(taskBytes), "{{", "}}")
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 
 	newTaskStr, err := common.Replace(fstTmpl, woc.globalParams.Merge(scope.getParameters()), true)
 	if err != nil {
@@ -595,7 +598,10 @@ func expandTask(task wfv1.DAGTask) ([]wfv1.DAGTask, error) {
 		return []wfv1.DAGTask{task}, nil
 	}
 
-	fstTmpl := fasttemplate.New(string(taskBytes), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(taskBytes), "{{", "}}")
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	expandedTasks := make([]wfv1.DAGTask, 0)
 	for i, item := range items {
 		var newTask wfv1.DAGTask
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 6b49e8f57..b84ed8280 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -2590,7 +2590,10 @@ func (woc *wfOperationCtx) substituteParamsInVolumes(params map[string]string) e
 	if err != nil {
 		return errors.InternalWrapError(err)
 	}
-	fstTmpl := fasttemplate.New(string(volumesBytes), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(volumesBytes), "{{", "}}")
+	if err != nil {
+		return fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	newVolumesStr, err := common.Replace(fstTmpl, params, true)
 	if err != nil {
 		return err
@@ -2659,7 +2662,11 @@ func (woc *wfOperationCtx) computeMetrics(metricList []*wfv1.Prometheus, localSc
 			woc.reportMetricEmissionError(fmt.Sprintf("unable to substitute parameters for metric '%s' (marshal): %s", metricTmpl.Name, err))
 			continue
 		}
-		fstTmpl := fasttemplate.New(string(metricTmplBytes), "{{", "}}")
+		fstTmpl, err := fasttemplate.NewTemplate(string(metricTmplBytes), "{{", "}}")
+		if err != nil {
+			woc.reportMetricEmissionError(fmt.Sprintf("unable to parse argo varaible for metric '%s': %s", metricTmpl.Name, err))
+			continue
+		}
 		replacedValue, err := common.Replace(fstTmpl, localScope, false)
 		if err != nil {
 			woc.reportMetricEmissionError(fmt.Sprintf("unable to substitute parameters for metric '%s': %s", metricTmpl.Name, err))
@@ -2715,7 +2722,11 @@ func (woc *wfOperationCtx) computeMetrics(metricList []*wfv1.Prometheus, localSc
 			metricSpec := metricTmpl.DeepCopy()
 
 			// Finally substitute value parameters
-			fstTmpl = fasttemplate.New(metricSpec.GetValueString(), "{{", "}}")
+			fstTmpl, err = fasttemplate.NewTemplate(metricSpec.GetValueString(), "{{", "}}")
+			if err != nil {
+				woc.reportMetricEmissionError(fmt.Sprintf("unable to parse argo varaible for metric '%s': %s", metricTmpl.Name, err))
+				continue
+			}
 			replacedValue, err := common.Replace(fstTmpl, localScope, false)
 			if err != nil {
 				woc.reportMetricEmissionError(fmt.Sprintf("unable to substitute parameters for metric '%s': %s", metricSpec.Name, err))
diff --git a/workflow/controller/steps.go b/workflow/controller/steps.go
index 70bbd2b28..c2d24ef22 100644
--- a/workflow/controller/steps.go
+++ b/workflow/controller/steps.go
@@ -348,7 +348,10 @@ func (woc *wfOperationCtx) resolveReferences(stepGroup []wfv1.WorkflowStep, scop
 		if err != nil {
 			return nil, errors.InternalWrapError(err)
 		}
-		fstTmpl := fasttemplate.New(string(stepBytes), "{{", "}}")
+		fstTmpl, err := fasttemplate.NewTemplate(string(stepBytes), "{{", "}}")
+		if err != nil {
+			return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+		}
 
 		newStepStr, err := common.Replace(fstTmpl, woc.globalParams.Merge(scope.getParameters()), true)
 		if err != nil {
@@ -431,7 +434,10 @@ func (woc *wfOperationCtx) expandStep(step wfv1.WorkflowStep) ([]wfv1.WorkflowSt
 	if err != nil {
 		return nil, errors.InternalWrapError(err)
 	}
-	fstTmpl := fasttemplate.New(string(stepBytes), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(stepBytes), "{{", "}}")
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	expandedStep := make([]wfv1.WorkflowStep, 0)
 	var items []wfv1.Item
 	if len(step.WithItems) > 0 {
diff --git a/workflow/controller/workflowpod.go b/workflow/controller/workflowpod.go
index 8e456f2f7..bbaaf42e4 100644
--- a/workflow/controller/workflowpod.go
+++ b/workflow/controller/workflowpod.go
@@ -357,7 +357,10 @@ func substitutePodParams(pod *apiv1.Pod, globalParams common.Parameters, tmpl *w
 	if err != nil {
 		return nil, err
 	}
-	fstTmpl := fasttemplate.New(string(specBytes), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(specBytes), "{{", "}}")
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	newSpecBytes, err := common.Replace(fstTmpl, podParams, true)
 	if err != nil {
 		return nil, err
@@ -1111,7 +1114,10 @@ func verifyResolvedVariables(obj interface{}) error {
 		return err
 	}
 	var unresolvedErr error
-	fstTmpl := fasttemplate.New(string(str), "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(string(str), "{{", "}}")
+	if err != nil {
+		return fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 	fstTmpl.ExecuteFuncString(func(w io.Writer, tag string) (int, error) {
 		unresolvedErr = errors.Errorf(errors.CodeBadRequest, "failed to resolve {{%s}}", tag)
 		return 0, nil
diff --git a/workflow/util/util.go b/workflow/util/util.go
index d768880f2..16ac41021 100644
--- a/workflow/util/util.go
+++ b/workflow/util/util.go
@@ -546,15 +546,16 @@ func FormulateResubmitWorkflow(wf *wfv1.Workflow, memoized bool) (*wfv1.Workflow
 	newWF.Spec.Shutdown = ""
 
 	// carry over user labels and annotations from previous workflow.
-	// skip any argoproj.io labels except for the controller instanceID label.
 	if newWF.ObjectMeta.Labels == nil {
 		newWF.ObjectMeta.Labels = make(map[string]string)
 	}
 	for key, val := range wf.ObjectMeta.Labels {
-		if strings.HasPrefix(key, workflow.WorkflowFullName+"/") && key != common.LabelKeyControllerInstanceID {
-			continue
+		switch key {
+		case common.LabelKeyCreator, common.LabelKeyPhase, common.LabelKeyCompleted:
+			// ignore
+		default:
+			newWF.ObjectMeta.Labels[key] = val
 		}
-		newWF.ObjectMeta.Labels[key] = val
 	}
 	// Append an additional label so it's easy for user to see the
 	// name of the original workflow that has been resubmitted.
diff --git a/workflow/validate/validate.go b/workflow/validate/validate.go
index 634d99e9a..35c94369c 100644
--- a/workflow/validate/validate.go
+++ b/workflow/validate/validate.go
@@ -89,23 +89,6 @@ var (
 	placeholderGenerator = common.NewPlaceholderGenerator()
 )
 
-// Allowed WfSpec fields if workflow referred WorkflowTemplate reference.
-var wfTmplRefAllowedWfSpecValidFields = map[string]bool{
-	"Entrypoint":            true,
-	"Suspend":               true,
-	"ActiveDeadlineSeconds": true,
-	"Priority":              true,
-	"Arguments":             true,
-	"WorkflowTemplateRef":   true,
-	"TTLStrategy":           true,
-	"Parallelism":           true,
-	"Volumes":               true,
-	"VolumeClaimTemplates":  true,
-	"NodeSelector":          true,
-	"PodGC":                 true,
-	"ServiceAccountName":    true,
-}
-
 type FakeArguments struct{}
 
 func (args *FakeArguments) GetParameterByName(name string) *wfv1.Parameter {
@@ -134,7 +117,7 @@ func ValidateWorkflow(wftmplGetter templateresolution.WorkflowTemplateNamespaced
 	hasWorkflowTemplateRef := wf.Spec.WorkflowTemplateRef != nil
 
 	if hasWorkflowTemplateRef {
-		err := ValidateWorkflowSpecFields(wf.Spec, wfTmplRefAllowedWfSpecValidFields)
+		err := ValidateWorkflowTemplateRefFields(wf.Spec)
 		if err != nil {
 			return nil, err
 		}
@@ -271,17 +254,9 @@ func ValidateWorkflow(wftmplGetter templateresolution.WorkflowTemplateNamespaced
 	return wfConditions, nil
 }
 
-func ValidateWorkflowSpecFields(v interface{}, validFieldMap map[string]bool) error {
-	val := reflect.ValueOf(v)
-
-	for i := 0; i < val.NumField(); i++ {
-		// Lookup the validate tag
-		field := val.Type().Field(i)
-		fieldVal := val.Field(i).IsZero()
-		_, ok := validFieldMap[field.Name]
-		if !ok && !fieldVal {
-			return errors.Errorf(errors.CodeBadRequest, "%s is invalid field in spec if workflow referred WorkflowTemplate reference", field.Name)
-		}
+func ValidateWorkflowTemplateRefFields(wfSpec wfv1.WorkflowSpec) error {
+	if len(wfSpec.Templates) > 0 {
+		return errors.Errorf(errors.CodeBadRequest, "Templates is invalid field in spec if workflow referred WorkflowTemplate reference")
 	}
 	return nil
 }
@@ -614,7 +589,10 @@ func resolveAllVariables(scope map[string]interface{}, tmplStr string) error {
 	_, allowAllItemRefs := scope[anyItemMagicValue] // 'item.*' is a magic placeholder value set by addItemsToScope
 	_, allowAllWorkflowOutputParameterRefs := scope[anyWorkflowOutputParameterMagicValue]
 	_, allowAllWorkflowOutputArtifactRefs := scope[anyWorkflowOutputArtifactMagicValue]
-	fstTmpl := fasttemplate.New(tmplStr, "{{", "}}")
+	fstTmpl, err := fasttemplate.NewTemplate(tmplStr, "{{", "}}")
+	if err != nil {
+		return fmt.Errorf("unable to parse argo varaible: %w", err)
+	}
 
 	fstTmpl.ExecuteFuncString(func(w io.Writer, tag string) (int, error) {
 


Test patch:
diff --git a/test/e2e/argo_server_test.go b/test/e2e/argo_server_test.go
index 654c1e2ad..4200986da 100644
--- a/test/e2e/argo_server_test.go
+++ b/test/e2e/argo_server_test.go
@@ -23,7 +23,7 @@ import (
 	"github.com/argoproj/argo/test/e2e/fixtures"
 )
 
-const baseUrl = "https://localhost:2746"
+const baseUrl = "http://localhost:2746"
 
 var httpClient = &http.Client{
 	Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
diff --git a/test/e2e/cli_with_server_test.go b/test/e2e/cli_with_server_test.go
index 8da62df4d..21f772220 100644
--- a/test/e2e/cli_with_server_test.go
+++ b/test/e2e/cli_with_server_test.go
@@ -25,8 +25,7 @@ func (s *CLIWithServerSuite) BeforeTest(suiteName, testName string) {
 	token, err := s.GetServiceAccountToken()
 	s.CheckError(err)
 	_ = os.Setenv("ARGO_SERVER", "localhost:2746")
-	_ = os.Setenv("ARGO_SECURE", "true")
-	_ = os.Setenv("ARGO_INSECURE_SKIP_VERIFY", "true")
+	_ = os.Setenv("ARGO_SECURE", "false")
 	_ = os.Setenv("ARGO_TOKEN", "Bearer "+token)
 	// we should not need this to run any tests
 	s.kubeConfig = os.Getenv("KUBECONFIG")
@@ -37,7 +36,6 @@ func (s *CLIWithServerSuite) AfterTest(suiteName, testName string) {
 	_ = os.Setenv("KUBECONFIG", s.kubeConfig)
 	_ = os.Unsetenv("ARGO_SERVER")
 	_ = os.Unsetenv("ARGO_SECURE")
-	_ = os.Unsetenv("ARGO_INSECURE_SKIP_VERIFY")
 	_ = os.Unsetenv("ARGO_TOKEN")
 	s.CLISuite.AfterTest(suiteName, testName)
 }
diff --git a/util/util_test.go b/util/util_test.go
index ae32030a4..a4b9b7bf3 100644
--- a/util/util_test.go
+++ b/util/util_test.go
@@ -37,23 +37,30 @@ func TestRecoverWorkflowNameFromSelectorString(t *testing.T) {
 	}{
 		{"TestRecoverWorkflowNameFromSelectorString", args{"metadata.name=whalesay"}, "whalesay"},
 		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"metadata.name="}, ""},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"metadata.name=whalesay,other=hello"}, "whalesay"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,metadata.name=whalesay,other=hello"}, "whalesay"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,metadata.name=whalesay"}, "whalesay"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,metadata.name= whalesay ,other=hello"}, "whalesay"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,other=hello"}, ""},
+		{"TestRecoverWorkflowNameFromSelectorString", args{"metadata.name=@latest"}, "@latest"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"metadata.name="}, ""},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"metadata.name=@latest,other=hello"}, "@latest"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,metadata.name=@latest,other=hello"}, "@latest"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,metadata.name=@latest"}, "@latest"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,metadata.name= @latest ,other=hello"}, "@latest"},
+		{"TestRecoverWorkflowNameFromSelectorStringEmptyWf", args{"foo=bar,other=hello"}, ""},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := RecoverWorkflowNameFromSelectorString(tt.args.selector)
-			assert.NoError(t, err)
+			got := RecoverWorkflowNameFromSelectorStringIfAny(tt.args.selector)
 			if got != tt.want {
-				t.Errorf("RecoverWorkflowNameFromSelectorString() = %v, want %v", got, tt.want)
+				t.Errorf("RecoverWorkflowNameFromSelectorStringIfAny() = %v, want %v", got, tt.want)
 			}
 		})
 	}
-}
-
-func TestRecoverWorkflowNameFromSelectorStringError(t *testing.T) {
-	name, err := RecoverWorkflowNameFromSelectorString("whatever=whalesay")
-	assert.NotNil(t, err)
+	name := RecoverWorkflowNameFromSelectorStringIfAny("whatever=whalesay")
 	assert.Equal(t, name, "")
 	assert.NotPanics(t, func() {
-		_, _ = RecoverWorkflowNameFromSelectorString("whatever")
+		_ = RecoverWorkflowNameFromSelectorStringIfAny("whatever")
 	})
 }
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 1328fca09..af6b55caa 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -4482,15 +4482,13 @@ metadata:
   name: steps-9fvnv
   namespace: argo
 spec:
-  activeDeadlineSeconds: 1
+  activeDeadlineSeconds: 3
   entrypoint: main
   templates:
   - name: main
     steps:
     - - name: approve
         template: approve
-      - name: hello1
-        template: whalesay
   - name: approve
     suspend: {}
   - container:
@@ -4520,25 +4518,11 @@ status:
       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
@@ -4573,7 +4557,7 @@ func TestFailSuspendedAndPendingNodesAfterDeadline(t *testing.T) {
 		woc.operate()
 		assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
 	})
-	time.Sleep(1 * time.Second)
+	time.Sleep(3 * time.Second)
 	t.Run("After Deadline", func(t *testing.T) {
 		woc.operate()
 		assert.Equal(t, wfv1.NodeFailed, woc.wf.Status.Phase)
diff --git a/workflow/util/util_test.go b/workflow/util/util_test.go
index 4a6a798f6..fbf7ad5b9 100644
--- a/workflow/util/util_test.go
+++ b/workflow/util/util_test.go
@@ -15,6 +15,7 @@ import (
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	fakeClientset "github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
+	"github.com/argoproj/argo/workflow/common"
 	hydratorfake "github.com/argoproj/argo/workflow/hydrator/fake"
 )
 
@@ -521,3 +522,33 @@ func TestApplySubmitOpts(t *testing.T) {
 		}
 	})
 }
+
+func TestFormulateResubmitWorkflow(t *testing.T) {
+	t.Run("Labels", func(t *testing.T) {
+		wf := &wfv1.Workflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					common.LabelKeyControllerInstanceID:    "1",
+					common.LabelKeyClusterWorkflowTemplate: "1",
+					common.LabelKeyCronWorkflow:            "1",
+					common.LabelKeyWorkflowTemplate:        "1",
+					common.LabelKeyCreator:                 "1",
+					common.LabelKeyPhase:                   "1",
+					common.LabelKeyCompleted:               "1",
+				},
+			},
+		}
+		wf, err := FormulateResubmitWorkflow(wf, false)
+		if assert.NoError(t, err) {
+			assert.Contains(t, wf.GetLabels(), common.LabelKeyControllerInstanceID)
+			assert.Contains(t, wf.GetLabels(), common.LabelKeyClusterWorkflowTemplate)
+			assert.Contains(t, wf.GetLabels(), common.LabelKeyCronWorkflow)
+			assert.Contains(t, wf.GetLabels(), common.LabelKeyWorkflowTemplate)
+			assert.NotContains(t, wf.GetLabels(), common.LabelKeyCreator)
+			assert.NotContains(t, wf.GetLabels(), common.LabelKeyPhase)
+			assert.NotContains(t, wf.GetLabels(), common.LabelKeyCompleted)
+			assert.Contains(t, wf.GetLabels(), common.LabelKeyPreviousWorkflowName)
+		}
+	})
+
+}