Support Priority Fields in Workflow Pods

api/openapi-spec/swagger.json

@@ -947,6 +947,15 @@
           "type": "integer",
           "format": "int64"
         },
+        "priority": {
+          "description": "Priority to apply to workflow pods.",
+          "type": "integer",
+          "format": "int32"
+        },
+        "priorityClassName": {
+          "description": "PriorityClassName to apply to workflow pods.",
+          "type": "string"
+        },
         "resource": {
           "description": "Resource template subtype which can run k8s resources",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ResourceTemplate"
@@ -1122,6 +1131,15 @@
           "type": "integer",
           "format": "int64"
         },
+        "podPriority": {
+          "description": "Priority to apply to workflow pods.",
+          "type": "integer",
+          "format": "int32"
+        },
+        "podPriorityClassName": {
+          "description": "PriorityClassName to apply to workflow pods.",
+          "type": "string"
+        },
         "priority": {
           "description": "Priority is used if controller is configured to process limited number of workflows in parallel. Workflows with higher priority are processed first.",
           "type": "integer",

pkg/apis/workflow/v1alpha1/types.go

@@ -145,6 +145,7 @@ type WorkflowSpec struct {
 	// allowed to run before the controller terminates the workflow. A value of zero is used to
 	// terminate a Running workflow
 	ActiveDeadlineSeconds *int64 `json:"activeDeadlineSeconds,omitempty"`
+
 	// Priority is used if controller is configured to process limited number of workflows in parallel. Workflows with higher priority are processed first.
 	Priority *int32 `json:"priority,omitempty"`
 
@@ -153,6 +154,12 @@ type WorkflowSpec struct {
 	// Default scheduler will be used if neither specified.
 	// +optional
 	SchedulerName string `json:"schedulerName,omitempty"`
+
+	// PriorityClassName to apply to workflow pods.
+	PodPriorityClassName string `json:"podPriorityClassName,omitempty"`
+
+	// Priority to apply to workflow pods.
+	PodPriority *int32 `json:"podPriority,omitempty"`
 }
 
 // Template is a reusable and composable unit of execution in a workflow
@@ -229,6 +236,12 @@ type Template struct {
 	// If neither specified, the pod will be dispatched by default scheduler.
 	// +optional
 	SchedulerName string `json:"schedulerName,omitempty"`
+
+	// PriorityClassName to apply to workflow pods.
+	PriorityClassName string `json:"priorityClassName,omitempty"`
+
+	// Priority to apply to workflow pods.
+	Priority *int32 `json:"priority,omitempty"`
 }
 
 // Inputs are the mechanism for passing parameters, artifacts, volumes from one template to another

workflow/controller/workflowpod.go

@@ -408,12 +408,31 @@ func addSchedulingConstraints(pod *apiv1.Pod, wfSpec *wfv1.WorkflowSpec, tmpl *w
 	} else if len(wfSpec.Tolerations) > 0 {
 		pod.Spec.Tolerations = wfSpec.Tolerations
 	}
+
 	// Set scheduler name (if specified)
 	if tmpl.SchedulerName != "" {
 		pod.Spec.SchedulerName = tmpl.SchedulerName
 	} else if wfSpec.SchedulerName != "" {
 		pod.Spec.SchedulerName = wfSpec.SchedulerName
 	}
+	// Set priorityClass (if specified)
+	if tmpl.PriorityClassName != "" {
+		pod.Spec.PriorityClassName = tmpl.PriorityClassName
+	} else if wfSpec.PodPriorityClassName != "" {
+		pod.Spec.PriorityClassName = wfSpec.PodPriorityClassName
+	}
+	// Set priority (if specified)
+	if tmpl.Priority != nil {
+		pod.Spec.Priority = tmpl.Priority
+	} else if wfSpec.PodPriority != nil {
+		pod.Spec.Priority = wfSpec.PodPriority
+	}
+	// Set schedulerName (if specified)
+	if tmpl.SchedulerName != "" {
+		pod.Spec.SchedulerName = tmpl.SchedulerName
+	} else if wfSpec.SchedulerName != "" {
+		pod.Spec.SchedulerName = wfSpec.SchedulerName
+	}
 }
 
 // addVolumeReferences adds any volumeMounts that a container/sidecar is referencing, to the pod.spec.volumes

workflow/controller/workflowpod_test.go

@@ -316,3 +316,28 @@ func TestOutOfCluster(t *testing.T) {
 		assert.Equal(t, "--kubeconfig=/some/path/config", pod.Spec.Containers[1].Args[1])
 	}
 }
+
+// TestPriority verifies the ability to carry forward priorityClassName and priority.
+func TestPriority(t *testing.T) {
+	priority := int32(15)
+	woc := newWoc()
+	woc.wf.Spec.Templates[0].PriorityClassName = "foo"
+	woc.wf.Spec.Templates[0].Priority = &priority
+	woc.executeContainer(woc.wf.Spec.Entrypoint, &woc.wf.Spec.Templates[0], "")
+	podName := getPodName(woc.wf)
+	pod, err := woc.controller.kubeclientset.CoreV1().Pods("").Get(podName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, pod.Spec.PriorityClassName, "foo")
+	assert.Equal(t, pod.Spec.Priority, &priority)
+}
+
+// TestSchedulerName verifies the ability to carry forward schedulerName.
+func TestSchedulerName(t *testing.T) {
+	woc := newWoc()
+	woc.wf.Spec.Templates[0].SchedulerName = "foo"
+	woc.executeContainer(woc.wf.Spec.Entrypoint, &woc.wf.Spec.Templates[0], "")
+	podName := getPodName(woc.wf)
+	pod, err := woc.controller.kubeclientset.CoreV1().Pods("").Get(podName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, pod.Spec.SchedulerName, "foo")
+}