Allow users to set priority class via CDI cr. (#1896)

Allow users to specify the priority class of the CDI control
plane using the CDI CR. Verifies that the set value is valid
before passing to the deployments.

Signed-off-by: Alexander Wels <awels@redhat.com>

api/openapi-spec/swagger.json

@@ -3635,6 +3635,10 @@
       "default": {},
       "$ref": "#/definitions/api.NodePlacement"
      },
+     "priorityClass": {
+      "description": "PriorityClass of the CDI control plane",
+      "type": "string"
+     },
      "uninstallStrategy": {
       "description": "CDIUninstallStrategy defines the state to leave CDI on uninstall",
       "type": "string"

go.mod

@@ -82,6 +82,7 @@ replace (
 	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.20.2
 	k8s.io/sample-cli-plugin => k8s.io/sample-cli-plugin v0.20.2
 	k8s.io/sample-controller => k8s.io/sample-controller v0.20.2
+	k8s.io/schedule => k8s.io/schedule v0.20.2
 
 	sigs.k8s.io/structured-merge-diff => sigs.k8s.io/structured-merge-diff v1.0.0
 	vbom.ml/util => github.com/fvbommel/util v0.0.0-20180919145318-efcd4e0f9787

pkg/apis/core/v1beta1/openapi_generated.go

@@ -14682,6 +14682,13 @@ func schema_pkg_apis_core_v1beta1_CDISpec(ref common.ReferenceCallback) common.O
 							Ref:         ref("kubevirt.io/containerized-data-importer/pkg/apis/core/v1beta1.CDICertConfig"),
 						},
 					},
+					"priorityClass": {
+						SchemaProps: spec.SchemaProps{
+							Description: "PriorityClass of the CDI control plane",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
 				},
 			},
 		},

pkg/apis/core/v1beta1/types.go

@@ -567,8 +567,13 @@ type CDISpec struct {
 	Config *CDIConfigSpec `json:"config,omitempty"`
 	// certificate configuration
 	CertConfig *CDICertConfig `json:"certConfig,omitempty"`
+	// PriorityClass of the CDI control plane
+	PriorityClass *CDIPriorityClass `json:"priorityClass,omitempty"`
 }
 
+// CDIPriorityClass defines the priority class of the CDI control plane.
+type CDIPriorityClass string
+
 // CDICloneStrategy defines the preferred method for performing a CDI clone (override snapshot?)
 type CDICloneStrategy string
 

pkg/apis/core/v1beta1/types_swagger_generated.go

@@ -302,6 +302,7 @@ func (CDISpec) SwaggerDoc() map[string]string {
 		"cloneStrategyOverride": "Clone strategy override: should we use a host-assisted copy even if snapshots are available?\n+kubebuilder:validation:Enum=\"copy\";\"snapshot\"",
 		"config":                "CDIConfig at CDI level",
 		"certConfig":            "certificate configuration",
+		"priorityClass":         "PriorityClass of the CDI control plane",
 	}
 }
 

pkg/apis/core/v1beta1/zz_generated.deepcopy.go

@@ -291,6 +291,11 @@ func (in *CDISpec) DeepCopyInto(out *CDISpec) {
 		*out = new(CDICertConfig)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PriorityClass != nil {
+		in, out := &in.PriorityClass, &out.PriorityClass
+		*out = new(CDIPriorityClass)
+		**out = **in
+	}
 	return
 }
 

pkg/operator/controller/BUILD.bazel

@@ -37,6 +37,7 @@ go_library(
         "//vendor/k8s.io/api/apps/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/rbac/v1:go_default_library",
+        "//vendor/k8s.io/api/scheduling/v1:go_default_library",
         "//vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",

pkg/operator/controller/cr-manager.go

@@ -17,6 +17,8 @@ limitations under the License.
 package controller
 
 import (
+	"context"
+
 	cdicluster "kubevirt.io/containerized-data-importer/pkg/operator/resources/cluster"
 	cdinamespaced "kubevirt.io/containerized-data-importer/pkg/operator/resources/namespaced"
 
@@ -26,7 +28,9 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	schedulingv1 "k8s.io/api/scheduling/v1"
 	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/types"
 	apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
 
 	cdiv1 "kubevirt.io/containerized-data-importer/pkg/apis/core/v1beta1"
@@ -77,6 +81,17 @@ func (r *ReconcileCDI) getNamespacedArgs(cr *cdiv1.CDI) *cdinamespaced.FactoryAr
 		if cr.Spec.ImagePullPolicy != "" {
 			result.PullPolicy = string(cr.Spec.ImagePullPolicy)
 		}
+		if cr.Spec.PriorityClass != nil && string(*cr.Spec.PriorityClass) != "" {
+			result.PriorityClassName = string(*cr.Spec.PriorityClass)
+		} else {
+			result.PriorityClassName = "openshift-user-critical"
+		}
+		// Verify the priority class name exists.
+		priorityClass := &schedulingv1.PriorityClass{}
+		if err := r.client.Get(context.TODO(), types.NamespacedName{Name: result.PriorityClassName}, priorityClass); err != nil {
+			// Any error we cannot determine if priority class exists.
+			result.PriorityClassName = ""
+		}
 		result.InfraNodePlacement = &cr.Spec.Infra
 	}
 

pkg/operator/resources/cluster/controller.go

@@ -197,6 +197,19 @@ func getControllerClusterPolicyRules() []rbacv1.PolicyRule {
 				"watch",
 			},
 		},
+		{
+			APIGroups: []string{
+				"scheduling.k8s.io",
+			},
+			Resources: []string{
+				"priorityclasses",
+			},
+			Verbs: []string{
+				"get",
+				"list",
+				"watch",
+			},
+		},
 	}
 }
 

pkg/operator/resources/crds_generated.go

@@ -1466,6 +1466,9 @@ spec:
                       type: object
                     type: array
                 type: object
+              priorityClass:
+                description: PriorityClass of the CDI control plane
+                type: string
               uninstallStrategy:
                 description: CDIUninstallStrategy defines the state to leave CDI on uninstall
                 enum:

pkg/operator/resources/namespaced/apiserver.go

@@ -42,7 +42,7 @@ func createAPIServerResources(args *FactoryArgs) []client.Object {
 		createAPIServerRoleBinding(),
 		createAPIServerRole(),
 		createAPIServerService(),
-		createAPIServerDeployment(args.APIServerImage, args.Verbosity, args.PullPolicy, args.InfraNodePlacement),
+		createAPIServerDeployment(args.APIServerImage, args.Verbosity, args.PullPolicy, args.PriorityClassName, args.InfraNodePlacement),
 	}
 }
 
@@ -87,9 +87,12 @@ func createAPIServerService() *corev1.Service {
 	return service
 }
 
-func createAPIServerDeployment(image, verbosity, pullPolicy string, infraNodePlacement *sdkapi.NodePlacement) *appsv1.Deployment {
+func createAPIServerDeployment(image, verbosity, pullPolicy, priorityClassName string, infraNodePlacement *sdkapi.NodePlacement) *appsv1.Deployment {
 	defaultMode := corev1.ConfigMapVolumeSourceDefaultMode
 	deployment := utils.CreateDeployment(apiServerRessouceName, cdiLabel, apiServerRessouceName, apiServerRessouceName, 1, infraNodePlacement)
+	if priorityClassName != "" {
+		deployment.Spec.Template.Spec.PriorityClassName = priorityClassName
+	}
 	container := utils.CreateContainer(apiServerRessouceName, image, verbosity, pullPolicy)
 	container.ReadinessProbe = &corev1.Probe{
 		Handler: corev1.Handler{

pkg/operator/resources/namespaced/controller.go

@@ -47,6 +47,7 @@ func createControllerResources(args *FactoryArgs) []client.Object {
 			args.UploadServerImage,
 			args.Verbosity,
 			args.PullPolicy,
+			args.PriorityClassName,
 			args.InfraNodePlacement),
 		createInsecureRegConfigMap(),
 	}
@@ -90,9 +91,12 @@ func createControllerServiceAccount() *corev1.ServiceAccount {
 	return utils.ResourceBuilder.CreateServiceAccount(common.ControllerServiceAccountName)
 }
 
-func createControllerDeployment(controllerImage, importerImage, clonerImage, uploadServerImage, verbosity, pullPolicy string, infraNodePlacement *sdkapi.NodePlacement) *appsv1.Deployment {
+func createControllerDeployment(controllerImage, importerImage, clonerImage, uploadServerImage, verbosity, pullPolicy, priorityClassName string, infraNodePlacement *sdkapi.NodePlacement) *appsv1.Deployment {
 	defaultMode := corev1.ConfigMapVolumeSourceDefaultMode
 	deployment := utils.CreateDeployment(controllerResourceName, "app", "containerized-data-importer", common.ControllerServiceAccountName, int32(1), infraNodePlacement)
+	if priorityClassName != "" {
+		deployment.Spec.Template.Spec.PriorityClassName = priorityClassName
+	}
 	container := utils.CreateContainer("cdi-controller", controllerImage, verbosity, pullPolicy)
 	container.Env = []corev1.EnvVar{
 		{

pkg/operator/resources/namespaced/factory.go

@@ -38,6 +38,7 @@ type FactoryArgs struct {
 	UploadServerImage      string `required:"true" split_words:"true"`
 	Verbosity              string `required:"true"`
 	PullPolicy             string `required:"true" split_words:"true"`
+	PriorityClassName      string
 	Namespace              string
 	InfraNodePlacement     *sdkapi.NodePlacement
 }

pkg/operator/resources/namespaced/uploadproxy.go

@@ -37,7 +37,7 @@ func createUploadProxyResources(args *FactoryArgs) []client.Object {
 		createUploadProxyService(),
 		createUploadProxyRoleBinding(),
 		createUploadProxyRole(),
-		createUploadProxyDeployment(args.UploadProxyImage, args.Verbosity, args.PullPolicy, args.InfraNodePlacement),
+		createUploadProxyDeployment(args.UploadProxyImage, args.Verbosity, args.PullPolicy, args.PriorityClassName, args.InfraNodePlacement),
 	}
 }
 
@@ -82,9 +82,12 @@ func createUploadProxyRole() *rbacv1.Role {
 	return utils.ResourceBuilder.CreateRole(uploadProxyResourceName, rules)
 }
 
-func createUploadProxyDeployment(image, verbosity, pullPolicy string, infraNodePlacement *sdkapi.NodePlacement) *appsv1.Deployment {
+func createUploadProxyDeployment(image, verbosity, pullPolicy, priorityClassName string, infraNodePlacement *sdkapi.NodePlacement) *appsv1.Deployment {
 	defaultMode := corev1.ConfigMapVolumeSourceDefaultMode
 	deployment := utils.CreateDeployment(uploadProxyResourceName, cdiLabel, uploadProxyResourceName, uploadProxyResourceName, int32(1), infraNodePlacement)
+	if priorityClassName != "" {
+		deployment.Spec.Template.Spec.PriorityClassName = priorityClassName
+	}
 	container := utils.CreateContainer(uploadProxyResourceName, image, verbosity, pullPolicy)
 	container.Env = []corev1.EnvVar{
 		{

tests/BUILD.bazel

@@ -74,6 +74,7 @@ go_test(
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/networking/v1:go_default_library",
         "//vendor/k8s.io/api/rbac/v1:go_default_library",
+        "//vendor/k8s.io/api/scheduling/v1:go_default_library",
         "//vendor/k8s.io/api/storage/v1:go_default_library",
         "//vendor/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/equality:go_default_library",

tests/cdiconfig_test.go

@@ -293,7 +293,7 @@ var _ = Describe("CDI ingress config tests, using manifests", func() {
 
 	It("[test_id:4950]Keep current value on no http rule ingress", func() {
 		manifestFile = "manifests/ingressNoHttp.yaml"
-		controllerPod, err := utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, cdiDeploymentPodPrefix, "app=containerized-data-importer")
+		controllerPod, err := utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, cdiDeploymentPodPrefix, common.CDILabelSelector)
 		Expect(err).ToNot(HaveOccurred())
 		currentRestarts := controllerPod.Status.ContainerStatuses[0].RestartCount
 		fmt.Fprintf(GinkgoWriter, "INFO: Current number of restarts: %d\n", currentRestarts)
@@ -311,7 +311,7 @@ var _ = Describe("CDI ingress config tests, using manifests", func() {
 		for i := 0; i < 10; i++ {
 			// Check for 20 seconds if the deployment pod crashed.
 			time.Sleep(2 * time.Second)
-			controllerPod, err = utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, cdiDeploymentPodPrefix, "app=containerized-data-importer")
+			controllerPod, err = utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, cdiDeploymentPodPrefix, common.CDILabelSelector)
 			Expect(err).ToNot(HaveOccurred())
 			// Restarts will increase if we crashed due to null pointer.
 			Expect(currentRestarts).To(Equal(controllerPod.Status.ContainerStatuses[0].RestartCount))

tests/leaderelection_test.go

@@ -22,11 +22,13 @@ import (
 )
 
 const (
-	cdiDeploymentName      = "cdi-deployment"
-	cdiDeploymentPodPrefix = "cdi-deployment-"
-	cdiOperatorName        = "cdi-operator"
-	cdiOperatorPodPrefix   = "cdi-operator-"
-	newDeploymentName      = "cdi-new-deployment"
+	cdiDeploymentName       = "cdi-deployment"
+	cdiDeploymentPodPrefix  = "cdi-deployment-"
+	cdiApiServerPodPrefix   = "cdi-apiserver-"
+	cdiUploadProxyPodPrefix = "cdi-uploadproxy-"
+	cdiOperatorName         = "cdi-operator"
+	cdiOperatorPodPrefix    = "cdi-operator-"
+	newDeploymentName       = "cdi-new-deployment"
 
 	pollingInterval = 2 * time.Second
 	timeout         = 360 * time.Second
@@ -248,7 +250,7 @@ func cleanupTest(f *framework.Framework, newDeployment *appsv1.Deployment) {
 }
 
 func getDeployments(f *framework.Framework) *appsv1.DeploymentList {
-	deployments, err := f.K8sClient.AppsV1().Deployments(f.CdiInstallNs).List(context.TODO(), metav1.ListOptions{LabelSelector: "app=containerized-data-importer"})
+	deployments, err := f.K8sClient.AppsV1().Deployments(f.CdiInstallNs).List(context.TODO(), metav1.ListOptions{LabelSelector: common.CDILabelSelector})
 	Expect(err).ToNot(HaveOccurred())
 	return deployments
 }
@@ -260,7 +262,7 @@ func getOperatorDeployment(f *framework.Framework) *appsv1.Deployment {
 }
 
 func getPods(f *framework.Framework) *v1.PodList {
-	pods, err := f.K8sClient.CoreV1().Pods(f.CdiInstallNs).List(context.TODO(), metav1.ListOptions{LabelSelector: "app=containerized-data-importer"})
+	pods, err := f.K8sClient.CoreV1().Pods(f.CdiInstallNs).List(context.TODO(), metav1.ListOptions{LabelSelector: common.CDILabelSelector})
 	Expect(err).ToNot(HaveOccurred())
 	return pods
 }

tests/operator_test.go

@@ -17,6 +17,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	schedulev1 "k8s.io/api/scheduling/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -780,6 +781,119 @@ var _ = Describe("ALL Operator tests", func() {
 				}, 2*time.Minute, 1*time.Second).Should(BeTrue())
 			})
 		})
+
+		var _ = Describe("Priority class tests", func() {
+			var (
+				cdi                   *cdiv1.CDI
+				systemClusterCritical = cdiv1.CDIPriorityClass("system-cluster-critical")
+				osUserCrit            = &schedulev1.PriorityClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "openshift-user-critical",
+					},
+					Value: 10000,
+				}
+			)
+			f := framework.NewFramework("operator-priority-class-test")
+			verifyPodPriorityClass := func(prefix, priorityClassName, labelSelector string) {
+				Eventually(func() string {
+					controllerPod, err := utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, prefix, labelSelector)
+					if err != nil {
+						return ""
+					}
+					return controllerPod.Spec.PriorityClassName
+				}, 2*time.Minute, 1*time.Second).Should(BeEquivalentTo(priorityClassName))
+			}
+
+			BeforeEach(func() {
+				cr, err := f.CdiClient.CdiV1beta1().CDIs().Get(context.TODO(), "cdi", metav1.GetOptions{})
+				if errors.IsNotFound(err) {
+					Skip("CDI CR 'cdi' does not exist.  Probably managed by another operator so skipping.")
+				}
+				Expect(err).ToNot(HaveOccurred())
+				cdi = cr
+				if cr.Spec.PriorityClass != nil {
+					By(fmt.Sprintf("Current priority class is: [%s]", *cr.Spec.PriorityClass))
+				}
+			})
+
+			AfterEach(func() {
+				if cdi == nil {
+					return
+				}
+
+				cr, err := f.CdiClient.CdiV1beta1().CDIs().Get(context.TODO(), "cdi", metav1.GetOptions{})
+				Expect(err).ToNot(HaveOccurred())
+
+				cr.Spec.PriorityClass = cdi.Spec.PriorityClass
+
+				_, err = f.CdiClient.CdiV1beta1().CDIs().Update(context.TODO(), cr, metav1.UpdateOptions{})
+				Expect(err).ToNot(HaveOccurred())
+
+				if !utils.IsOpenshift(f.K8sClient) {
+					Eventually(func() bool {
+						return errors.IsNotFound(f.K8sClient.SchedulingV1().PriorityClasses().Delete(context.TODO(), osUserCrit.Name, metav1.DeleteOptions{}))
+					}, 2*time.Minute, 1*time.Second).Should(BeTrue())
+				}
+				By("Ensuring the CDI priority class is restored")
+				prioClass := ""
+				if cr.Spec.PriorityClass != nil {
+					prioClass = string(*cr.Spec.PriorityClass)
+				}
+				// Deployment
+				verifyPodPriorityClass(cdiDeploymentPodPrefix, string(prioClass), common.CDILabelSelector)
+				// API server
+				verifyPodPriorityClass(cdiApiServerPodPrefix, string(prioClass), "")
+				// Upload server
+				verifyPodPriorityClass(cdiUploadProxyPodPrefix, string(prioClass), "")
+				By("Verifying there is just a single cdi controller pod")
+				Eventually(func() error {
+					_, err := utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, cdiDeploymentPodPrefix, common.CDILabelSelector)
+					return err
+				}, 2*time.Minute, 1*time.Second).Should(BeNil())
+				By("Ensuring this pod is the leader")
+				Eventually(func() bool {
+					controllerPod, err := utils.FindPodByPrefix(f.K8sClient, f.CdiInstallNs, cdiDeploymentPodPrefix, common.CDILabelSelector)
+					Expect(err).ToNot(HaveOccurred())
+					log := getLog(f, controllerPod.Name)
+					return checkLogForRegEx(logIsLeaderRegex, log)
+				}, 2*time.Minute, 1*time.Second).Should(BeTrue())
+
+			})
+
+			It("should use kubernetes priority class if set", func() {
+				cr, err := f.CdiClient.CdiV1beta1().CDIs().Get(context.TODO(), "cdi", metav1.GetOptions{})
+				if errors.IsNotFound(err) {
+					Skip("CDI CR 'cdi' does not exist.  Probably managed by another operator so skipping.")
+				}
+				Expect(err).ToNot(HaveOccurred())
+				By("Setting the priority class to system cluster critical, which is known to exist")
+				cr.Spec.PriorityClass = &systemClusterCritical
+				_, err = f.CdiClient.CdiV1beta1().CDIs().Update(context.TODO(), cr, metav1.UpdateOptions{})
+				Expect(err).ToNot(HaveOccurred())
+				By("Verifying the CDI deployment is updated")
+				verifyPodPriorityClass(cdiDeploymentPodPrefix, string(systemClusterCritical), common.CDILabelSelector)
+				By("Verifying the CDI api server is updated")
+				verifyPodPriorityClass(cdiApiServerPodPrefix, string(systemClusterCritical), "")
+				By("Verifying the CDI upload proxy server is updated")
+				verifyPodPriorityClass(cdiUploadProxyPodPrefix, string(systemClusterCritical), "")
+			})
+
+			It("should use openshift priority class if not set and available", func() {
+				_, err := f.CdiClient.CdiV1beta1().CDIs().Get(context.TODO(), "cdi", metav1.GetOptions{})
+				if errors.IsNotFound(err) {
+					Skip("CDI CR 'cdi' does not exist.  Probably managed by another operator so skipping.")
+				}
+				_, err = f.K8sClient.SchedulingV1().PriorityClasses().Create(context.TODO(), osUserCrit, metav1.CreateOptions{})
+				Expect(err).ToNot(HaveOccurred())
+				By("Verifying the CDI control plane is updated")
+				// Deployment
+				verifyPodPriorityClass(cdiDeploymentPodPrefix, string(osUserCrit.Name), common.CDILabelSelector)
+				// API server
+				verifyPodPriorityClass(cdiApiServerPodPrefix, string(osUserCrit.Name), "")
+				// Upload server
+				verifyPodPriorityClass(cdiUploadProxyPodPrefix, string(osUserCrit.Name), "")
+			})
+		})
 	})
 })
 

tests/utils/pod.go

@@ -115,7 +115,7 @@ func findPodByCompFuncOnce(clientSet *kubernetes.Clientset, namespace, prefix, l
 	for _, pod := range podList.Items {
 		if compFunc(pod.Name, prefix) {
 			if result == nil {
-				result = &pod
+				result = pod.DeepCopy()
 			} else {
 				fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: First pod name %s in namespace %s\n", result.Name, result.Namespace)
 				fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: Second pod name %s in namespace %s\n", pod.Name, pod.Namespace)

vendor/modules.txt

@@ -1085,5 +1085,6 @@ sigs.k8s.io/yaml
 # k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.20.2
 # k8s.io/sample-cli-plugin => k8s.io/sample-cli-plugin v0.20.2
 # k8s.io/sample-controller => k8s.io/sample-controller v0.20.2
+# k8s.io/schedule => k8s.io/schedule v0.20.2
 # sigs.k8s.io/structured-merge-diff => sigs.k8s.io/structured-merge-diff v1.0.0
 # vbom.ml/util => github.com/fvbommel/util v0.0.0-20180919145318-efcd4e0f9787