fix: add workflow template level pod annotations and labels to templa…

…te. Fixes: #12945 (#12987)

Signed-off-by: shuangkun <[email protected]>

pkg/apis/workflow/v1alpha1/cluster_workflow_template_types.go

@@ -57,6 +57,11 @@ func (cwftmpl *ClusterWorkflowTemplate) GetResourceScope() ResourceScope {
 	return ResourceScopeCluster
 }
 
+// GetPodMetadata returns the PodMetadata of cluster workflow template.
+func (cwftmpl *ClusterWorkflowTemplate) GetPodMetadata() *Metadata {
+	return cwftmpl.Spec.PodMetadata
+}
+
 // GetWorkflowSpec returns the WorkflowSpec of cluster workflow template.
 func (cwftmpl *ClusterWorkflowTemplate) GetWorkflowSpec() *WorkflowSpec {
 	return &cwftmpl.Spec

pkg/apis/workflow/v1alpha1/common.go

@@ -20,6 +20,7 @@ type TemplateHolder interface {
 	GroupVersionKind() schema.GroupVersionKind
 	GetTemplateByName(name string) *Template
 	GetResourceScope() ResourceScope
+	GetPodMetadata() *Metadata
 }
 
 // WorkflowSpecHolder is an object that holds a WorkflowSpec; e.g., WorkflowTemplate, and ClusterWorkflowTemplate

pkg/apis/workflow/v1alpha1/workflow_template_types.go

@@ -56,6 +56,11 @@ func (wftmpl *WorkflowTemplate) GetResourceScope() ResourceScope {
 	return ResourceScopeNamespaced
 }
 
+// GetPodMetadata returns the PodMetadata of workflow template.
+func (wftmpl *WorkflowTemplate) GetPodMetadata() *Metadata {
+	return wftmpl.Spec.PodMetadata
+}
+
 // GetWorkflowSpec returns the WorkflowSpec of workflow template.
 func (wftmpl *WorkflowTemplate) GetWorkflowSpec() *WorkflowSpec {
 	return &wftmpl.Spec

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -3410,6 +3410,11 @@ func (wf *Workflow) GetResourceScope() ResourceScope {
 	return ResourceScopeLocal
 }
 
+// GetPodMetadata returns the PodMetadata of a workflow.
+func (wf *Workflow) GetPodMetadata() *Metadata {
+	return wf.Spec.PodMetadata
+}
+
 // GetWorkflowSpec returns the Spec of a workflow.
 func (wf *Workflow) GetWorkflowSpec() WorkflowSpec {
 	return wf.Spec

workflow/controller/operator_workflow_template_ref_test.go

@@ -633,3 +633,71 @@ func TestWorkflowTemplateWithDynamicRef(t *testing.T) {
 	woc.operate(ctx)
 	assert.Equal(t, wfv1.WorkflowSucceeded, woc.wf.Status.Phase)
 }
+
+const wfTemplateWithPodMetadata = `
+apiVersion: argoproj.io/v1alpha1
+kind: ClusterWorkflowTemplate
+metadata:
+  name: workflow-template
+spec:
+  entrypoint: whalesay-template
+  podMetadata:
+    labels:
+      workflow-template-label: hello
+    annotations:
+      all-pods-should-have-this: value
+  arguments:
+    parameters:
+      - name: message
+        value: hello world
+
+  templates:
+    - name: whalesay-template
+      inputs:
+        parameters:
+          - name: message
+      container:
+        image: docker/whalesay
+        command: [cowsay]
+        args: ["{{inputs.parameters.message}}"]`
+
+const wfWithTemplateRef = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: test-workflow
+  namespace: argo-workflows-system
+spec:
+  podMetadata:
+    labels:
+      caller-label: hello
+  entrypoint: start
+  templates:
+    - name: start
+      steps:
+        - - name: hello
+            templateRef:
+              name: workflow-template
+              template: whalesay-template
+              clusterScope: true
+            arguments:
+              parameters:
+                - name: message
+                  value: Hello Bug`
+
+func TestWorkflowTemplateWithPodMetadata(t *testing.T) {
+	cancel, controller := newController(wfv1.MustUnmarshalWorkflow(wfWithTemplateRef), wfv1.MustUnmarshalClusterWorkflowTemplate(wfTemplateWithPodMetadata))
+	defer cancel()
+
+	ctx := context.Background()
+	woc := newWorkflowOperationCtx(wfv1.MustUnmarshalWorkflow(wfWithTemplateRef), controller)
+	woc.operate(ctx)
+	assert.Equal(t, wfv1.WorkflowRunning, woc.wf.Status.Phase)
+	pods, err := listPods(woc)
+	require.NoError(t, err)
+	assert.NotEmpty(t, len(pods.Items) > 0, "pod was not created successfully")
+	pod := pods.Items[0]
+	assert.Contains(t, pod.Labels, "caller-label")
+	assert.Contains(t, pod.Labels, "workflow-template-label")
+	assert.Contains(t, pod.Annotations, "all-pods-should-have-this")
+}

workflow/templateresolution/context.go

@@ -104,6 +104,10 @@ func (ctx *Context) GetTemplateByName(name string) (*wfv1.Template, error) {
 	ctx.log.Debugf("Getting the template by name: %s", name)
 
 	tmpl := ctx.tmplBase.GetTemplateByName(name)
+
+	podMetadata := ctx.tmplBase.GetPodMetadata()
+	ctx.addPodMetadata(podMetadata, tmpl)
+
 	if tmpl == nil {
 		return nil, errors.Errorf(errors.CodeNotFound, "template %s not found", name)
 	}
@@ -138,6 +142,9 @@ func (ctx *Context) GetTemplateFromRef(tmplRef *wfv1.TemplateRef) (*wfv1.Templat
 
 	template = wftmpl.GetTemplateByName(tmplRef.Template)
 
+	podMetadata := wftmpl.GetPodMetadata()
+	ctx.addPodMetadata(podMetadata, template)
+
 	if template == nil {
 		return nil, errors.Errorf(errors.CodeNotFound, "template %s not found in workflow template %s", tmplRef.Template, tmplRef.Name)
 	}
@@ -268,3 +275,21 @@ func (ctx *Context) WithClusterWorkflowTemplate(name string) (*Context, error) {
 	}
 	return ctx.WithTemplateBase(cwftmpl), nil
 }
+
+// addPodMetadata add podMetadata in workflow template level to template
+func (ctx *Context) addPodMetadata(podMetadata *wfv1.Metadata, tmpl *wfv1.Template) {
+	if podMetadata != nil {
+		if tmpl.Metadata.Annotations == nil {
+			tmpl.Metadata.Annotations = make(map[string]string)
+		}
+		for k, v := range podMetadata.Annotations {
+			tmpl.Metadata.Annotations[k] = v
+		}
+		if tmpl.Metadata.Labels == nil {
+			tmpl.Metadata.Labels = make(map[string]string)
+		}
+		for k, v := range podMetadata.Labels {
+			tmpl.Metadata.Labels[k] = v
+		}
+	}
+}