Add an alert and an condition for RelieveAndMigrate without PSI

RelieveAndMigrate profile requires PSI metric to be
enabled for the worker nodes (psi=1 kernel parameter).
Directly check it for the node that is executing the
operator pod, looking if /proc/pressure/ is available
inside the pod, and report it with a condition.
Check it for all the worker nodes and report it also
with an alert.

Signed-off-by: Simone Tiraboschi <[email protected]>

Author: tiraboschi

README.md

@@ -180,6 +180,9 @@ relevant asymmetry between the descheduling and successive scheduling decisions.
 The soft taints set by the descheduler soft-tainter act as a hint for the scheduler to mitigate
 this asymmetry and foster a quicker convergence.
 
+This profile requires [PSI](https://docs.kernel.org/accounting/psi.html) metrics to be enabled (psi=1 kernel parameter)
+for all the worker nodes.
+
 The profile exposes the following customization:
 - `devLowNodeUtilizationThresholds`: Sets experimental thresholds for the LowNodeUtilization strategy.
 - `devActualUtilizationProfile`: Enable load-aware descheduling.

bindata/assets/kube-descheduler/psialert.yaml

@@ -0,0 +1,23 @@
+---
+apiVersion: monitoring.coreos.com/v1
+kind: PrometheusRule
+metadata:
+  name: descheduler-psi-alert
+  namespace: openshift-kube-descheduler-operator
+spec:
+  groups:
+    - name: recordingRules.alerts
+      rules:
+        - alert: DeschedulerPSIDisabled
+          expr: |-
+            count(kube_node_role{role="worker"}) > 
+            (count(
+              descheduler:nodepressure:cpu:avg1m * on (instance) group_left (node) label_replace(kube_node_role{role="worker"}, "instance", "$1", "node", "(.+)")) 
+              OR on() vector(0)
+            )
+          for: 0m
+          labels:
+            severity: critical
+          annotations:
+            summary: Kube Descheduler Operator is configured to consume a PSI metric but PSI is not enabled
+            description: "Kube Descheduler Operator is configured (devActualUtilizationProfile) to consume a PSI metric but PSI is not enabled for all the worker nodes (psi=1 kernel argument)"

pkg/operator/target_config_reconciler.go

@@ -3,6 +3,7 @@ package operator
 import (
 	"context"
 	"fmt"
+	"os"
 	"slices"
 	"strconv"
 	"strings"
@@ -67,6 +68,8 @@ import (
 
 const DefaultImage = "quay.io/openshift/origin-descheduler:latest"
 const kubeVirtShedulableLabelSelector = "kubevirt.io/schedulable=true"
+const psiPath = "/proc/pressure/"
+const EXPERIMENTAL_DISABLE_PSI_CHECK = "EXPERIMENTAL_DISABLE_PSI_CHECK"
 
 // deschedulerCommand provides descheduler command with policyconfigfile mounted as volume and log-level for backwards
 // compatibility with 3.11
@@ -88,6 +91,7 @@ type TargetConfigReconciler struct {
 	namespaceLister          corev1listers.NamespaceLister
 	nodeLister               corev1listers.NodeLister
 	cache                    resourceapply.ResourceCache
+	psiPath                  string
 }
 
 func NewTargetConfigReconciler(
@@ -133,6 +137,7 @@ func NewTargetConfigReconciler(
 		namespaceLister:          coreInformers.Core().V1().Namespaces().Lister(),
 		nodeLister:               coreInformers.Core().V1().Nodes().Lister(),
 		cache:                    resourceapply.NewResourceCache(),
+		psiPath:                  psiPath,
 	}
 
 	configInformer.Config().V1().Schedulers().Informer().AddEventHandler(c.eventHandler())
@@ -322,6 +327,16 @@ func (c TargetConfigReconciler) sync() error {
 		specAnnotations["clusterrolebindings/openshift-descheduler-operand"] = resourceVersion
 	}
 
+	if softtainterPSIAlert, _, err := c.managePSIAlert(descheduler, isSoftTainterNeeded); err != nil {
+		return err
+	} else {
+		resourceVersion := "0"
+		if softtainterPSIAlert != nil {
+			resourceVersion = softtainterPSIAlert.GetResourceVersion()
+		}
+		specAnnotations["prometheusrule/descheduler-psi-alert"] = resourceVersion
+	}
+
 	if role, _, err := c.manageRole(descheduler); err != nil {
 		return err
 	} else {
@@ -476,6 +491,22 @@ func (c *TargetConfigReconciler) managePrometheusRule(descheduler *deschedulerv1
 	return resourceapply.ApplyKnownUnstructured(c.ctx, c.dynamicClient, c.eventRecorder, required)
 }
 
+func (c *TargetConfigReconciler) managePSIAlert(descheduler *deschedulerv1.KubeDescheduler, stEnabled bool) (*unstructured.Unstructured, bool, error) {
+	required := resourceread.ReadUnstructuredOrDie(bindata.MustAsset("assets/kube-descheduler/psialert.yaml"))
+	ownerReference := metav1.OwnerReference{
+		APIVersion: "operator.openshift.io/v1",
+		Kind:       "KubeDescheduler",
+		Name:       descheduler.Name,
+		UID:        descheduler.UID,
+	}
+	required.SetOwnerReferences([]metav1.OwnerReference{ownerReference})
+	controller.EnsureOwnerRef(required, ownerReference)
+	if stEnabled {
+		return resourceapply.ApplyKnownUnstructured(c.ctx, c.dynamicClient, c.eventRecorder, required)
+	}
+	return resourceapply.DeleteKnownUnstructured(c.ctx, c.dynamicClient, c.eventRecorder, required)
+}
+
 func (c *TargetConfigReconciler) manageSoftTainterValidatingAdmissionPolicy(descheduler *deschedulerv1.KubeDescheduler, stEnabled bool) (*admissionv1.ValidatingAdmissionPolicy, bool, error) {
 	required := resourceread.ReadValidatingAdmissionPolicyV1OrDie(bindata.MustAsset("assets/kube-descheduler/softtaintervalidatingadmissionpolicy.yaml"))
 	ownerReference := metav1.OwnerReference{
@@ -1340,6 +1371,13 @@ func (c *TargetConfigReconciler) manageConfigMap(descheduler *deschedulerv1.Kube
 			if !kvDeployed {
 				return nil, true, fmt.Errorf("profile %v can only be used when KubeVirt is properly deployed", deschedulerv1.RelieveAndMigrate)
 			}
+			psiEnabled, psierr := c.isPSIenabled()
+			if psierr != nil {
+				return nil, false, psierr
+			}
+			if !psiEnabled {
+				return nil, true, fmt.Errorf("profile %v can only be used when PSI metrics are enabled for the worker nodes", deschedulerv1.RelieveAndMigrate)
+			}
 			kubeVirtShedulable := kubeVirtShedulableLabelSelector
 			policy.NodeSelector = &kubeVirtShedulable
 			profile, err = relieveAndMigrateProfile(descheduler.Spec.ProfileCustomizations, includedNamespaces, excludedNamespaces, c.protectedNamespaces)
@@ -1629,6 +1667,22 @@ func (c *TargetConfigReconciler) isKubeVirtDeployed() (bool, error) {
 	return false, nil
 }
 
+func (c *TargetConfigReconciler) isPSIenabled() (bool, error) {
+	if boolValue, err := strconv.ParseBool(os.Getenv(EXPERIMENTAL_DISABLE_PSI_CHECK)); err == nil && boolValue {
+		return true, nil
+	}
+
+	_, err := os.Stat(c.psiPath)
+	if err == nil {
+		return true, nil
+	} else {
+		if os.IsNotExist(err) {
+			return false, nil
+		}
+		return false, err
+	}
+}
+
 func (c *TargetConfigReconciler) isSoftTainterNeeded(descheduler *deschedulerv1.KubeDescheduler) (bool, error) {
 	if slices.Contains(descheduler.Spec.Profiles, deschedulerv1.RelieveAndMigrate) {
 		return true, nil

pkg/operator/target_config_reconciler_test.go

@@ -3,6 +3,8 @@ package operator
 import (
 	"context"
 	"fmt"
+	"os"
+	"path"
 	"testing"
 	"time"
 	"unsafe"
@@ -107,6 +109,17 @@ func TestManageConfigMap(t *testing.T) {
 	fiveMinutes := metav1.Duration{Duration: fm}
 	priority := int32(1000)
 
+	tempPSIPath, err := os.MkdirTemp("", "unittest")
+	if err != nil {
+		t.Fatalf("Failed test: %v", err)
+	}
+	defer func(tempPSIPath string) {
+		err := os.RemoveAll(tempPSIPath)
+		if err != nil {
+			t.Fatalf("Failed test: %v", err)
+		}
+	}(tempPSIPath)
+
 	tests := []struct {
 		name            string
 		schedulerConfig *configv1.Scheduler
@@ -116,6 +129,7 @@ func TestManageConfigMap(t *testing.T) {
 		nodes           []runtime.Object
 		err             error
 		forceDeployment bool
+		missingPSI      bool
 	}{
 		{
 			name: "Podlifetime",
@@ -583,6 +597,36 @@ func TestManageConfigMap(t *testing.T) {
 			err:             fmt.Errorf("profile DevKubeVirtRelieveAndMigrate can only be used when KubeVirt is properly deployed"),
 			forceDeployment: true,
 		},
+		{
+			name: "RelieveAndMigrateWithoutPSI",
+			descheduler: &deschedulerv1.KubeDescheduler{
+				Spec: deschedulerv1.KubeDeschedulerSpec{
+					Profiles:              []deschedulerv1.DeschedulerProfile{"DevKubeVirtRelieveAndMigrate"},
+					ProfileCustomizations: &deschedulerv1.ProfileCustomizations{DevLowNodeUtilizationThresholds: &deschedulerv1.LowThreshold},
+				},
+			},
+			want: &corev1.ConfigMap{
+				TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "ConfigMap"},
+				Data:     map[string]string{"policy.yaml": string(bindata.MustAsset("assets/relieveAndMigrateLowConfig.yaml"))},
+			},
+			nodes: []runtime.Object{
+				&corev1.Node{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:   "node1",
+						Labels: map[string]string{"kubevirt.io/schedulable": "true"},
+					},
+				},
+				&corev1.Node{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:   "node2",
+						Labels: map[string]string{"kubevirt.io/schedulable": "true"},
+					},
+				},
+			},
+			missingPSI:      true,
+			err:             fmt.Errorf("profile DevKubeVirtRelieveAndMigrate can only be used when PSI metrics are enabled for the worker nodes"),
+			forceDeployment: true,
+		},
 		{
 			name: "AffinityAndTaintsWithNamespaces",
 			descheduler: &deschedulerv1.KubeDescheduler{
@@ -889,7 +933,13 @@ func TestManageConfigMap(t *testing.T) {
 			ctx, cancelFunc := context.WithCancel(context.TODO())
 			defer cancelFunc()
 
+			testPSIPath := tempPSIPath
+			if tt.missingPSI {
+				testPSIPath = path.Join(tempPSIPath, "MISSING")
+			}
+
 			targetConfigReconciler, _ := initTargetConfigReconciler(ctx, objects, []runtime.Object{tt.schedulerConfig}, tt.routes, nil)
+			targetConfigReconciler.psiPath = testPSIPath
 
 			got, forceDeployment, err := targetConfigReconciler.manageConfigMap(tt.descheduler)
 			if tt.err != nil {
@@ -1100,6 +1150,18 @@ func TestManageDeployment(t *testing.T) {
 func TestManageSoftTainterDeployment(t *testing.T) {
 	ctx, cancelFunc := context.WithCancel(context.TODO())
 	defer cancelFunc()
+
+	tempPSIPath, err := os.MkdirTemp("", "unittest")
+	if err != nil {
+		t.Fatalf("Failed test: %v", err)
+	}
+	defer func(path string) {
+		err := os.RemoveAll(path)
+		if err != nil {
+			t.Fatalf("Failed test: %v", err)
+		}
+	}(tempPSIPath)
+
 	expectedSoftTainterDeployment := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "Deployment"},
 		ObjectMeta: metav1.ObjectMeta{
@@ -1387,6 +1449,7 @@ func TestManageSoftTainterDeployment(t *testing.T) {
 
 func TestSync(t *testing.T) {
 	fakeRecorder := NewFakeRecorder(1024)
+
 	tests := []struct {
 		name                   string
 		targetConfigReconciler *TargetConfigReconciler

test/e2e/operator_test.go

@@ -59,6 +59,8 @@ var operatorConfigs = map[string]string{
 }
 var operatorConfigsAppliers = map[string]func() error{}
 
+const EXPERIMENTAL_DISABLE_PSI_CHECK = "EXPERIMENTAL_DISABLE_PSI_CHECK"
+
 func TestMain(m *testing.M) {
 	if os.Getenv("KUBECONFIG") == "" {
 		klog.Errorf("KUBECONFIG environment variable not set")
@@ -265,6 +267,24 @@ func TestSoftTainterDeployment(t *testing.T) {
 		}
 	}(ctx, kubeClient)
 
+	// patch the operator deployment to mock PSI
+	prevDisablePSIcheck, foundDisablePSIcheck := os.LookupEnv(EXPERIMENTAL_DISABLE_PSI_CHECK)
+	if err := mockPSIEnv(ctx, kubeClient); err != nil {
+		t.Fatalf("Failed mocking PSI path enviromental variable to the operator depoyment: %v", err)
+	}
+	defer func(ctx context.Context, kubeClient *k8sclient.Clientset, prevDisablePSIcheck string, foundDisablePSIcheck bool) {
+		err := unmockPSIEnv(ctx, kubeClient, prevDisablePSIcheck, foundDisablePSIcheck)
+		if err != nil {
+			t.Fatalf("Failed PSI path enviromental variable: %v", err)
+		}
+	}(ctx, kubeClient, prevDisablePSIcheck, foundDisablePSIcheck)
+	// wait for descheduler operator pod to be running
+	deschOpPod, err := waitForPodRunningByNamePrefix(ctx, kubeClient, operatorclient.OperatorNamespace, operatorclient.OperandName, operatorclient.OperandName+"-operator")
+	if err != nil {
+		t.Fatalf("Unable to wait for the Descheduler operator pod to run")
+	}
+	klog.Infof("Descheduler pod running in %v", deschOpPod.Name)
+
 	// apply devKubeVirtRelieveAndMigrate CR for the operator
 	if err := operatorConfigsAppliers[kubeVirtRelieveAndMigrateConf](); err != nil {
 		t.Fatalf("Unable to apply a CR for Descheduler operator: %v", err)
@@ -358,6 +378,24 @@ func TestSoftTainterVAP(t *testing.T) {
 		}
 	}(ctx, kubeClient)
 
+	// patch the operator deployment to mock PSI
+	prevDisablePSIcheck, foundDisablePSIcheck := os.LookupEnv(EXPERIMENTAL_DISABLE_PSI_CHECK)
+	if err := mockPSIEnv(ctx, kubeClient); err != nil {
+		t.Fatalf("Failed mocking PSI path enviromental variable to the operator depoyment: %v", err)
+	}
+	defer func(ctx context.Context, kubeClient *k8sclient.Clientset, prevDisablePSIcheck string, foundDisablePSIcheck bool) {
+		err := unmockPSIEnv(ctx, kubeClient, prevDisablePSIcheck, foundDisablePSIcheck)
+		if err != nil {
+			t.Fatalf("Failed PSI path enviromental variable: %v", err)
+		}
+	}(ctx, kubeClient, prevDisablePSIcheck, foundDisablePSIcheck)
+	// wait for descheduler operator pod to be running
+	deschOpPod, err := waitForPodRunningByNamePrefix(ctx, kubeClient, operatorclient.OperatorNamespace, operatorclient.OperandName, operatorclient.OperandName+"-operator")
+	if err != nil {
+		t.Fatalf("Unable to wait for the Descheduler operator pod to run")
+	}
+	klog.Infof("Descheduler pod running in %v", deschOpPod.Name)
+
 	// apply devKubeVirtRelieveAndMigrate CR for the operator
 	if err := operatorConfigsAppliers[kubeVirtRelieveAndMigrateConf](); err != nil {
 		t.Fatalf("Unable to apply a CR for Descheduler operator: %v", err)
@@ -825,6 +863,27 @@ func updateNodeAndRetryOnConflicts(ctx context.Context, kubeClient *k8sclient.Cl
 	return nil
 }
 
+func updateDeploymentAndRetryOnConflicts(ctx context.Context, kubeClient *k8sclient.Clientset, deployment *appsv1.Deployment, opts metav1.UpdateOptions) error {
+	uDeployment, uerr := kubeClient.AppsV1().Deployments(deployment.Namespace).Update(ctx, deployment, opts)
+	if uerr != nil {
+		if apierrors.IsConflict(uerr) {
+			if uDeployment.Name == "" {
+				uDeployment, uerr = kubeClient.AppsV1().Deployments(uDeployment.Namespace).Get(ctx, uDeployment.Name, metav1.GetOptions{})
+				if uerr != nil {
+					return uerr
+				}
+			}
+			deployment.Spec.DeepCopyInto(&uDeployment.Spec)
+			uDeployment.ObjectMeta.Labels = deployment.ObjectMeta.Labels
+			uDeployment.ObjectMeta.Annotations = deployment.ObjectMeta.Annotations
+			_, err := kubeClient.AppsV1().Deployments(deployment.Namespace).Update(ctx, deployment, opts)
+			return err
+		}
+		return uerr
+	}
+	return nil
+}
+
 func tryUpdatingTaintWithExpectation(ctx context.Context, t *testing.T, clientSet *k8sclient.Clientset, node *corev1.Node, taint *corev1.Taint, add, expectedSuccess bool) {
 	rnode, err := clientSet.CoreV1().Nodes().Get(ctx, node.Name, metav1.GetOptions{})
 	if err != nil {
@@ -886,3 +945,40 @@ func tryRemovingTaintWithExpectedSuccess(ctx context.Context, t *testing.T, clie
 func tryRemovingTaintWithExpectedFailure(ctx context.Context, t *testing.T, clientSet *k8sclient.Clientset, node *corev1.Node, taint *corev1.Taint) {
 	tryUpdatingTaintWithExpectation(ctx, t, clientSet, node, taint, false, false)
 }
+
+func mockPSIEnv(ctx context.Context, kubeClient *k8sclient.Clientset) error {
+	operatorDeployment, err := kubeClient.AppsV1().Deployments(operatorclient.OperatorNamespace).Get(ctx, operatorclient.OperandName+"-operator", metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+	operatorDeployment.Spec.Template.Spec.Containers[0].Env = append(
+		operatorDeployment.Spec.Template.Spec.Containers[0].Env,
+		v1.EnvVar{
+			Name:  EXPERIMENTAL_DISABLE_PSI_CHECK,
+			Value: "true",
+		})
+	return updateDeploymentAndRetryOnConflicts(ctx, kubeClient, operatorDeployment, metav1.UpdateOptions{})
+}
+
+func unmockPSIEnv(ctx context.Context, kubeClient *k8sclient.Clientset, prevDisablePSIcheck string, foundDisablePSIcheck bool) error {
+	operatorDeployment, err := kubeClient.AppsV1().Deployments(operatorclient.OperatorNamespace).Get(ctx, operatorclient.OperandName+"-operator", metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+	var envVars []v1.EnvVar
+	for _, e := range operatorDeployment.Spec.Template.Spec.Containers[0].Env {
+		if e.Name != EXPERIMENTAL_DISABLE_PSI_CHECK {
+			envVars = append(envVars, e)
+		}
+	}
+	if foundDisablePSIcheck {
+		operatorDeployment.Spec.Template.Spec.Containers[0].Env = append(
+			envVars,
+			v1.EnvVar{
+				Name:  EXPERIMENTAL_DISABLE_PSI_CHECK,
+				Value: prevDisablePSIcheck,
+			})
+	}
+	operatorDeployment.Spec.Template.Spec.Containers[0].Env = envVars
+	return updateDeploymentAndRetryOnConflicts(ctx, kubeClient, operatorDeployment, metav1.UpdateOptions{})
+}