Merge pull request #368 from arangodb/feature/pod-priority

Feature/pod priority

Author: Lars Maier

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [0.3.10](---) (XXXX-XX-XX)
 - Added Pod Disruption Budgets for all server groups in production mode.
+- Added Priority Class Name to be specified per server group.
 
 ## [0.3.9](https://github.com/arangodb/kube-arangodb/tree/0.3.9) (2019-02-28)
 [Full Changelog](https://github.com/arangodb/kube-arangodb/compare/0.3.8...0.3.9)

docs/Manual/Deployment/Kubernetes/DeploymentResource.md

@@ -413,6 +413,10 @@ for each server of this group.
 This setting is not available for group `coordinators`, `syncmasters` & `syncworkers`
 because servers in these groups do not need persistent storage.
 
+### `spec.<group>.priorityClassName: string`
+
+Priority class name for pods of this group. Will be forwarded to the pod spec. [Kubernetes documentation](https://kubernetes.io/docs/concepts/configuration/pod-priority-preemption/)
+
 ### `spec.<group>.probes.livenessProbeDisabled: bool`
 
 If set to true, the operator does not generate a liveness probe for new pods belonging to this group.

manifests/templates/test/rbac.yaml

@@ -15,6 +15,9 @@ rules:
 - apiGroups: ["apps"]
   resources: ["daemonsets", "deployments"]
   verbs: ["*"]
+- apiGroups: ["scheduling.k8s.io"]
+  resources: ["priorityclasses"]
+  verbs: ["*"]
 
 ---
 

pkg/apis/deployment/v1alpha/deployment_status_members.go

@@ -243,3 +243,23 @@ func (ds DeploymentStatusMembers) AllMembersReady(mode DeploymentMode, syncEnabl
 		return false
 	}
 }
+
+// MembersOfGroup returns the member list of the given group
+func (ds DeploymentStatusMembers) MembersOfGroup(group ServerGroup) MemberStatusList {
+	switch group {
+	case ServerGroupSingle:
+		return ds.Single
+	case ServerGroupAgents:
+		return ds.Agents
+	case ServerGroupDBServers:
+		return ds.DBServers
+	case ServerGroupCoordinators:
+		return ds.Coordinators
+	case ServerGroupSyncMasters:
+		return ds.SyncMasters
+	case ServerGroupSyncWorkers:
+		return ds.SyncWorkers
+	default:
+		return MemberStatusList{}
+	}
+}

pkg/apis/deployment/v1alpha/server_group_spec.go

@@ -58,6 +58,8 @@ type ServerGroupSpec struct {
 	NodeSelector map[string]string `json:"nodeSelector,omitempty"`
 	// Probes specifies additional behaviour for probes
 	Probes *ServerGroupProbesSpec `json:"probes,omitempty"`
+	// PriorityClassName specifies a priority class name
+	PriorityClassName string `json:"priorityClassName,omitempty"`
 }
 
 // ServerGroupProbesSpec contains specification for probes for pods of the server group

pkg/deployment/images.go

@@ -198,7 +198,7 @@ func (ib *imagesBuilder) fetchArangoDBImageIDAndVersion(ctx context.Context, ima
 		}
 	}
 	if err := k8sutil.CreateArangodPod(ib.KubeCli, true, ib.APIObject, role, id, podName, "", image, "", "", ib.Spec.GetImagePullPolicy(), "", false, terminationGracePeriod, args, env, nil, nil, nil,
-		tolerations, serviceAccountName, "", "", "", nil, v1.ResourceRequirements{}); err != nil {
+		tolerations, serviceAccountName, "", "", "", nil, "", v1.ResourceRequirements{}); err != nil {
 		log.Debug().Err(err).Msg("Failed to create image ID pod")
 		return true, maskAny(err)
 	}

pkg/deployment/reconcile/plan_builder.go

@@ -345,9 +345,15 @@ func podNeedsRotation(log zerolog.Logger, p v1.Pod, apiObject metav1.Object, spe
 		return true, "ServiceAccountName changed"
 	}
 
+	// Check priorities
+	if groupSpec.PriorityClassName != p.Spec.PriorityClassName {
+		return true, "Pod priority changed"
+	}
+
 	// Check resource requirements
 	if resourcesRequireRotation(spec.GetServerGroupSpec(group).Resources, k8sutil.GetArangoDBContainerFromPod(&p).Resources) {
 		return true, "Resource Requirements changed"
+
 	}
 
 	return false, ""

pkg/deployment/resources/pod_creator.go

@@ -607,7 +607,7 @@ func (r *Resources) createPodForMember(spec api.DeploymentSpec, memberID string,
 		finalizers := r.createPodFinalizers(group)
 		if err := k8sutil.CreateArangodPod(kubecli, spec.IsDevelopment(), apiObject, role, m.ID, m.PodName, m.PersistentVolumeClaimName, imageInfo.ImageID, lifecycleImage, alpineImage, spec.GetImagePullPolicy(),
 			engine, requireUUID, terminationGracePeriod, args, env, finalizers, livenessProbe, readinessProbe, tolerations, serviceAccountName, tlsKeyfileSecretName, rocksdbEncryptionSecretName,
-			clusterJWTSecretName, groupSpec.GetNodeSelector(), groupSpec.Resources); err != nil {
+			clusterJWTSecretName, groupSpec.GetNodeSelector(), groupSpec.PriorityClassName, groupSpec.Resources); err != nil {
 			return maskAny(err)
 		}
 		log.Debug().Str("pod-name", m.PodName).Msg("Created pod")
@@ -689,7 +689,8 @@ func (r *Resources) createPodForMember(spec api.DeploymentSpec, memberID string,
 			affinityWithRole = api.ServerGroupDBServers.AsRole()
 		}
 		if err := k8sutil.CreateArangoSyncPod(kubecli, spec.IsDevelopment(), apiObject, role, m.ID, m.PodName, imageID, lifecycleImage, spec.GetImagePullPolicy(), terminationGracePeriod, args, env,
-			livenessProbe, tolerations, serviceAccountName, tlsKeyfileSecretName, clientAuthCASecretName, masterJWTSecretName, clusterJWTSecretName, affinityWithRole, groupSpec.GetNodeSelector()); err != nil {
+			livenessProbe, tolerations, serviceAccountName, tlsKeyfileSecretName, clientAuthCASecretName, masterJWTSecretName, clusterJWTSecretName, affinityWithRole, groupSpec.GetNodeSelector(),
+			groupSpec.PriorityClassName, groupSpec.Resources); err != nil {
 			return maskAny(err)
 		}
 		log.Debug().Str("pod-name", m.PodName).Msg("Created pod")

pkg/util/k8sutil/pods.go

@@ -321,7 +321,7 @@ func arangodContainer(image string, imagePullPolicy v1.PullPolicy, args []string
 
 // arangosyncContainer creates a container configured to run `arangosync`.
 func arangosyncContainer(image string, imagePullPolicy v1.PullPolicy, args []string, env map[string]EnvValue, livenessProbe *HTTPProbeConfig,
-	lifecycle *v1.Lifecycle, lifecycleEnvVars []v1.EnvVar) v1.Container {
+	lifecycle *v1.Lifecycle, lifecycleEnvVars []v1.EnvVar, resources v1.ResourceRequirements) v1.Container {
 	c := v1.Container{
 		Command:         append([]string{"/usr/sbin/arangosync"}, args...),
 		Name:            ServerContainerName,
@@ -335,6 +335,7 @@ func arangosyncContainer(image string, imagePullPolicy v1.PullPolicy, args []str
 				Protocol:      v1.ProtocolTCP,
 			},
 		},
+		Resources: filterStorageResourceRequirement(resources), // Storage is handled via pvcs
 	}
 	for k, v := range env {
 		c.Env = append(c.Env, v.CreateEnvVar(k))
@@ -456,7 +457,8 @@ func CreateArangodPod(kubecli kubernetes.Interface, developmentMode bool, deploy
 	args []string, env map[string]EnvValue, finalizers []string,
 	livenessProbe *HTTPProbeConfig, readinessProbe *HTTPProbeConfig, tolerations []v1.Toleration, serviceAccountName string,
 	tlsKeyfileSecretName, rocksdbEncryptionSecretName string, clusterJWTSecretName string, nodeSelector map[string]string,
-	resources v1.ResourceRequirements) error {
+	podPriorityClassName string, resources v1.ResourceRequirements) error {
+
 	// Prepare basic pod
 	p := newPod(deployment.GetName(), deployment.GetNamespace(), role, id, podName, finalizers, tolerations, serviceAccountName, nodeSelector)
 	terminationGracePeriodSeconds := int64(math.Ceil(terminationGracePeriod.Seconds()))
@@ -491,6 +493,9 @@ func CreateArangodPod(kubecli kubernetes.Interface, developmentMode bool, deploy
 	}
 	p.Spec.Containers = append(p.Spec.Containers, c)
 
+	// Add priorityClassName
+	p.Spec.PriorityClassName = podPriorityClassName
+
 	// Add UUID init container
 	if alpineImage != "" {
 		p.Spec.InitContainers = append(p.Spec.InitContainers, arangodInitContainer("uuid", id, engine, alpineImage, requireUUID))
@@ -575,7 +580,8 @@ func CreateArangodPod(kubecli kubernetes.Interface, developmentMode bool, deploy
 // If another error occurs, that error is returned.
 func CreateArangoSyncPod(kubecli kubernetes.Interface, developmentMode bool, deployment APIObject, role, id, podName, image, lifecycleImage string, imagePullPolicy v1.PullPolicy,
 	terminationGracePeriod time.Duration, args []string, env map[string]EnvValue, livenessProbe *HTTPProbeConfig, tolerations []v1.Toleration, serviceAccountName string,
-	tlsKeyfileSecretName, clientAuthCASecretName, masterJWTSecretName, clusterJWTSecretName, affinityWithRole string, nodeSelector map[string]string) error {
+	tlsKeyfileSecretName, clientAuthCASecretName, masterJWTSecretName, clusterJWTSecretName, affinityWithRole string, nodeSelector map[string]string,
+	podPriorityClassName string, resources v1.ResourceRequirements) error {
 	// Prepare basic pod
 	p := newPod(deployment.GetName(), deployment.GetNamespace(), role, id, podName, nil, tolerations, serviceAccountName, nodeSelector)
 	terminationGracePeriodSeconds := int64(math.Ceil(terminationGracePeriod.Seconds()))
@@ -601,7 +607,7 @@ func CreateArangoSyncPod(kubecli kubernetes.Interface, developmentMode bool, dep
 	p.Spec.Volumes = append(p.Spec.Volumes, lifecycleVolumes...)
 
 	// Add arangosync container
-	c := arangosyncContainer(image, imagePullPolicy, args, env, livenessProbe, lifecycle, lifecycleEnvVars)
+	c := arangosyncContainer(image, imagePullPolicy, args, env, livenessProbe, lifecycle, lifecycleEnvVars, resources)
 	if tlsKeyfileSecretName != "" {
 		c.VolumeMounts = append(c.VolumeMounts, tlsKeyfileVolumeMounts()...)
 	}
@@ -616,6 +622,9 @@ func CreateArangoSyncPod(kubecli kubernetes.Interface, developmentMode bool, dep
 	}
 	p.Spec.Containers = append(p.Spec.Containers, c)
 
+	// Add priorityClassName
+	p.Spec.PriorityClassName = podPriorityClassName
+
 	// TLS keyfile secret mount (if any)
 	if tlsKeyfileSecretName != "" {
 		vol := v1.Volume{

tests/pc_test.go

@@ -0,0 +1,126 @@
+package tests
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
+	"github.com/arangodb/kube-arangodb/pkg/client"
+	"github.com/arangodb/kube-arangodb/pkg/util"
+	"github.com/arangodb/kube-arangodb/pkg/util/retry"
+	"github.com/dchest/uniuri"
+	v1beta1 "k8s.io/api/scheduling/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/arangodb/kube-arangodb/pkg/generated/clientset/versioned"
+)
+
+func waitForPriorityOfServerGroup(kube kubernetes.Interface, c versioned.Interface, depl, ns string, group api.ServerGroup, priority int32) error {
+	return retry.Retry(func() error {
+
+		apiObject, err := c.DatabaseV1alpha().ArangoDeployments(ns).Get(depl, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+
+		for _, m := range apiObject.Status.Members.MembersOfGroup(group) {
+			pod, err := kube.CoreV1().Pods(apiObject.Namespace).Get(m.PodName, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+
+			if pod.Spec.Priority == nil {
+				return fmt.Errorf("No pod priority set")
+			}
+
+			if *pod.Spec.Priority != priority {
+				return fmt.Errorf("Wrong pod priority, expected %d, found %d", priority, pod.Spec.Priority)
+			}
+		}
+
+		return nil
+	}, 5*time.Minute)
+}
+
+// TestPriorityClasses creates a PriorityClass and associates coordinators with that class.
+// Then check if the pods have the desired priority. Then change the class and check that the pods are rotated.
+func TestPriorityClasses(t *testing.T) {
+	longOrSkip(t)
+	c := client.MustNewInCluster()
+	kubecli := mustNewKubeClient(t)
+	ns := getNamespace(t)
+
+	lowClassName := "test-low-class"
+	lowClassValue := int32(1000)
+
+	highClassName := "test-high-class"
+	highClassValue := int32(2000)
+
+	// Create two priority classes
+	if _, err := kubecli.SchedulingV1beta1().PriorityClasses().Create(&v1beta1.PriorityClass{
+		Value:         lowClassValue,
+		GlobalDefault: false,
+		Description:   "Low priority test class",
+		ObjectMeta: metav1.ObjectMeta{
+			Name: lowClassName,
+		},
+	}); err != nil {
+		t.Fatalf("Could not create PC: %v", err)
+	}
+	defer kubecli.SchedulingV1beta1().PriorityClasses().Delete(lowClassName, &metav1.DeleteOptions{})
+
+	if _, err := kubecli.SchedulingV1beta1().PriorityClasses().Create(&v1beta1.PriorityClass{
+		Value:         highClassValue,
+		GlobalDefault: false,
+		Description:   "Low priority test class",
+		ObjectMeta: metav1.ObjectMeta{
+			Name: highClassName,
+		},
+	}); err != nil {
+		t.Fatalf("Could not create PC: %v", err)
+	}
+	defer kubecli.SchedulingV1beta1().PriorityClasses().Delete(highClassName, &metav1.DeleteOptions{})
+
+	// Prepare deployment config
+	depl := newDeployment("test-pc-" + uniuri.NewLen(4))
+	depl.Spec.Mode = api.NewMode(api.DeploymentModeCluster)
+	depl.Spec.TLS = api.TLSSpec{CASecretName: util.NewString("None")}
+	depl.Spec.DBServers.Count = util.NewInt(2)
+	depl.Spec.Coordinators.Count = util.NewInt(2)
+	depl.Spec.Coordinators.PriorityClassName = lowClassName
+	depl.Spec.SetDefaults(depl.GetName()) // this must be last
+	defer deferedCleanupDeployment(c, depl.GetName(), ns)
+
+	// Create deployment
+	_, err := c.DatabaseV1alpha().ArangoDeployments(ns).Create(depl)
+	if err != nil {
+		t.Fatalf("Create deployment failed: %v", err)
+	}
+	defer deferedCleanupDeployment(c, depl.GetName(), ns)
+
+	// Wait for deployment to be ready
+	_, err = waitUntilDeployment(c, depl.GetName(), ns, deploymentIsReady())
+	if err != nil {
+		t.Fatalf("Deployment not running in time: %v", err)
+	}
+
+	if err := waitForPriorityOfServerGroup(kubecli, c, depl.GetName(), ns, api.ServerGroupCoordinators, lowClassValue); err != nil {
+		t.Errorf("PDBs not as expected: %v", err)
+	}
+
+	_, err = updateDeployment(c, depl.GetName(), ns, func(spec *api.DeploymentSpec) {
+		spec.Coordinators.PriorityClassName = highClassName
+	})
+	if err != nil {
+		t.Fatalf("Failed to update deployment: %v", err)
+	}
+
+	// Check if priority class is updated
+	if err := waitForPriorityOfServerGroup(kubecli, c, depl.GetName(), ns, api.ServerGroupCoordinators, highClassValue); err != nil {
+		t.Errorf("Priority not as expected: %v", err)
+	}
+
+	removeDeployment(c, depl.GetName(), ns)
+}