Merge pull request #722 from Dentrax/feature/leaderelection

feat(leaderelection): impl leader election for HA Deployment

test/e2e/e2e_duplicatepods_test.go

@@ -20,16 +20,12 @@ import (
 	"context"
 	"strings"
 	"testing"
-	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
-	clientset "k8s.io/client-go/kubernetes"
-
 	deschedulerapi "sigs.k8s.io/descheduler/pkg/api"
 	"sigs.k8s.io/descheduler/pkg/descheduler/evictions"
 	eutils "sigs.k8s.io/descheduler/pkg/descheduler/evictions/utils"
@@ -109,7 +105,8 @@ func TestRemoveDuplicates(t *testing.T) {
 						Name: "sample",
 						VolumeSource: v1.VolumeSource{
 							EmptyDir: &v1.EmptyDirVolumeSource{
-								SizeLimit: resource.NewQuantity(int64(10), resource.BinarySI)},
+								SizeLimit: resource.NewQuantity(int64(10), resource.BinarySI),
+							},
 						},
 					},
 				}
@@ -177,27 +174,3 @@ func TestRemoveDuplicates(t *testing.T) {
 		})
 	}
 }
-
-func waitForPodsRunning(ctx context.Context, t *testing.T, clientSet clientset.Interface, labelMap map[string]string, desireRunningPodNum int, namespace string) {
-	if err := wait.PollImmediate(10*time.Second, 60*time.Second, func() (bool, error) {
-		podList, err := clientSet.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
-			LabelSelector: labels.SelectorFromSet(labelMap).String(),
-		})
-		if err != nil {
-			return false, err
-		}
-		if len(podList.Items) != desireRunningPodNum {
-			t.Logf("Waiting for %v pods to be running, got %v instead", desireRunningPodNum, len(podList.Items))
-			return false, nil
-		}
-		for _, pod := range podList.Items {
-			if pod.Status.Phase != v1.PodRunning {
-				t.Logf("Pod %v not running yet, is %v instead", pod.Name, pod.Status.Phase)
-				return false, nil
-			}
-		}
-		return true, nil
-	}); err != nil {
-		t.Fatalf("Error waiting for pods running: %v", err)
-	}
-}

test/e2e/e2e_leaderelection_test.go

@@ -0,0 +1,201 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	clientset "k8s.io/client-go/kubernetes"
+
+	"sigs.k8s.io/descheduler/cmd/descheduler/app/options"
+	"sigs.k8s.io/descheduler/pkg/descheduler"
+)
+
+func TestLeaderElection(t *testing.T) {
+	ctx := context.Background()
+
+	clientSet, _, _, stopCh := initializeClient(t)
+	defer close(stopCh)
+
+	ns1 := "e2e-" + strings.ToLower(t.Name()+"-a")
+	ns2 := "e2e-" + strings.ToLower(t.Name()+"-b")
+
+	t.Logf("Creating testing namespace %v", ns1)
+	testNamespace1 := &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ns1}}
+	if _, err := clientSet.CoreV1().Namespaces().Create(ctx, testNamespace1, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("Unable to create ns %v", testNamespace1.Name)
+	}
+	defer clientSet.CoreV1().Namespaces().Delete(ctx, testNamespace1.Name, metav1.DeleteOptions{})
+
+	t.Logf("Creating testing namespace %v", ns2)
+	testNamespace2 := &v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ns2}}
+	if _, err := clientSet.CoreV1().Namespaces().Create(ctx, testNamespace2, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("Unable to create ns %v", testNamespace2.Name)
+	}
+	defer clientSet.CoreV1().Namespaces().Delete(ctx, testNamespace2.Name, metav1.DeleteOptions{})
+
+	deployment1, err := createDeployment(ctx, clientSet, ns1, 5, t)
+	if err != nil {
+		t.Fatalf("create deployment 1: %v", err)
+	}
+	defer clientSet.AppsV1().Deployments(deployment1.Namespace).Delete(ctx, deployment1.Name, metav1.DeleteOptions{})
+
+	deployment2, err := createDeployment(ctx, clientSet, ns2, 5, t)
+	if err != nil {
+		t.Fatalf("create deployment 2: %v", err)
+	}
+	defer clientSet.AppsV1().Deployments(deployment2.Namespace).Delete(ctx, deployment2.Name, metav1.DeleteOptions{})
+
+	waitForPodsRunning(ctx, t, clientSet, map[string]string{"test": "leaderelection", "name": "test-leaderelection"}, 5, ns1)
+
+	podListAOrg := getPodNameList(ctx, clientSet, ns1, t)
+
+	waitForPodsRunning(ctx, t, clientSet, map[string]string{"test": "leaderelection", "name": "test-leaderelection"}, 5, ns2)
+
+	podListBOrg := getPodNameList(ctx, clientSet, ns2, t)
+
+	s1, err := options.NewDeschedulerServer()
+	if err != nil {
+		t.Fatalf("unable to initialize server: %v", err)
+	}
+	s1.Client = clientSet
+	s1.DeschedulingInterval = 5 * time.Second
+	s1.LeaderElection.LeaderElect = true
+	s1.KubeconfigFile = os.Getenv("KUBECONFIG")
+	s1.PolicyConfigFile = "./policy_leaderelection_a.yaml"
+
+	s2, err := options.NewDeschedulerServer()
+	if err != nil {
+		t.Fatalf("unable to initialize server: %v", err)
+	}
+	s2.Client = clientSet
+	s2.DeschedulingInterval = 5 * time.Second
+	s2.LeaderElection.LeaderElect = true
+	s2.KubeconfigFile = os.Getenv("KUBECONFIG")
+	s2.PolicyConfigFile = "./policy_leaderelection_b.yaml"
+
+	t.Log("starting deschedulers")
+
+	go func() {
+		err := descheduler.Run(ctx, s1)
+		if err != nil {
+			t.Errorf("unable to start descheduler: %v", err)
+			return
+		}
+	}()
+
+	time.Sleep(1 * time.Second)
+
+	go func() {
+		err := descheduler.Run(ctx, s2)
+		if err != nil {
+			t.Errorf("unable to start descheduler: %v", err)
+			return
+		}
+	}()
+
+	defer clientSet.CoordinationV1().Leases(s1.LeaderElection.ResourceNamespace).Delete(ctx, s1.LeaderElection.ResourceName, metav1.DeleteOptions{})
+	defer clientSet.CoordinationV1().Leases(s2.LeaderElection.ResourceNamespace).Delete(ctx, s2.LeaderElection.ResourceName, metav1.DeleteOptions{})
+
+	// wait for a while so all the pods are 5 seconds older
+	time.Sleep(7 * time.Second)
+
+	// validate only pods from e2e-testleaderelection-a namespace are evicted.
+	podListA := getPodNameList(ctx, clientSet, ns1, t)
+
+	podListB := getPodNameList(ctx, clientSet, ns2, t)
+
+	left := reflect.DeepEqual(podListAOrg, podListA)
+	right := reflect.DeepEqual(podListBOrg, podListB)
+
+	singleNamespaceEvicted := (left && !right) || (!left && right)
+
+	if singleNamespaceEvicted {
+		if !left {
+			t.Logf("Only the pods in %s namespace are evicted. Pods before: %s, Pods after %s", ns1, podListAOrg, podListA)
+		} else {
+			t.Logf("Only the pods in %s namespace are evicted. Pods before: %s, Pods after %s", ns2, podListBOrg, podListB)
+		}
+	} else {
+		t.Fatalf("Pods are evicted in both namespaces. For %s namespace Pods before: %s, Pods after %s. And, for %s namespace Pods before: %s, Pods after: %s", ns1, podListAOrg, podListA, ns2, podListBOrg, podListB)
+	}
+}
+
+func createDeployment(ctx context.Context, clientSet clientset.Interface, namespace string, replicas int32, t *testing.T) (*appsv1.Deployment, error) {
+	deployment := &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "leaderelection",
+			Namespace: namespace,
+			Labels:    map[string]string{"test": "leaderelection", "name": "test-leaderelection"},
+		},
+		Spec: appsv1.DeploymentSpec{
+			Replicas: func(i int32) *int32 { return &i }(replicas),
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{"test": "leaderelection", "name": "test-leaderelection"},
+			},
+			Template: v1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{"test": "leaderelection", "name": "test-leaderelection"},
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{{
+						Name:            "pause",
+						ImagePullPolicy: "Always",
+						Image:           "kubernetes/pause",
+						Ports:           []v1.ContainerPort{{ContainerPort: 80}},
+					}},
+				},
+			},
+		},
+	}
+
+	t.Logf("Creating deployment %v for namespace %s", deployment.Name, deployment.Namespace)
+	deployment, err := clientSet.AppsV1().Deployments(deployment.Namespace).Create(ctx, deployment, metav1.CreateOptions{})
+	if err != nil {
+		t.Logf("Error creating deployment: %v", err)
+		if err = clientSet.AppsV1().Deployments(deployment.Namespace).DeleteCollection(ctx, metav1.DeleteOptions{}, metav1.ListOptions{
+			LabelSelector: labels.SelectorFromSet(labels.Set(map[string]string{"test": "leaderelection", "name": "test-leaderelection"})).String(),
+		}); err != nil {
+			t.Fatalf("Unable to delete deployment: %v", err)
+		}
+		return nil, fmt.Errorf("create deployment %v", err)
+	}
+	return deployment, nil
+}
+
+func getPodNameList(ctx context.Context, clientSet clientset.Interface, namespace string, t *testing.T) []string {
+	podList, err := clientSet.CoreV1().Pods(namespace).List(
+		ctx, metav1.ListOptions{LabelSelector: labels.SelectorFromSet(labels.Set(map[string]string{"test": "leaderelection", "name": "test-leaderelection"})).String()})
+	if err != nil {
+		t.Fatalf("Unable to list pods from ns: %s: %v", namespace, err)
+	}
+	podNames := make([]string, len(podList.Items))
+	for i, pod := range podList.Items {
+		podNames[i] = pod.Name
+	}
+	return podNames
+}

test/e2e/e2e_test.go

@@ -517,8 +517,8 @@ func TestEvictSystemCriticalPriorityClass(t *testing.T) {
 }
 
 func testEvictSystemCritical(t *testing.T, isPriorityClass bool) {
-	var highPriority = int32(1000)
-	var lowPriority = int32(500)
+	highPriority := int32(1000)
+	lowPriority := int32(500)
 	ctx := context.Background()
 
 	clientSet, nodeInformer, getPodsAssignedToNode, stopCh := initializeClient(t)
@@ -647,8 +647,8 @@ func TestThresholdPriorityClass(t *testing.T) {
 }
 
 func testPriority(t *testing.T, isPriorityClass bool) {
-	var highPriority = int32(1000)
-	var lowPriority = int32(500)
+	highPriority := int32(1000)
+	lowPriority := int32(500)
 	ctx := context.Background()
 
 	clientSet, nodeInformer, getPodsAssignedToNode, stopCh := initializeClient(t)
@@ -745,7 +745,7 @@ func testPriority(t *testing.T, isPriorityClass bool) {
 		t.Fatalf("None of %v high priority pods are expected to be deleted", expectReservePodNames)
 	}
 
-	//check if all pods with low priority class are evicted
+	// check if all pods with low priority class are evicted
 	if err := wait.PollImmediate(time.Second, 30*time.Second, func() (bool, error) {
 		podListLowPriority, err := clientSet.CoreV1().Pods(rcLowPriority.Namespace).List(
 			ctx, metav1.ListOptions{LabelSelector: labels.SelectorFromSet(rcLowPriority.Spec.Template.Labels).String()})
@@ -848,7 +848,7 @@ func TestPodLabelSelector(t *testing.T) {
 		t.Fatalf("None of %v unevictable pods are expected to be deleted", expectReservePodNames)
 	}
 
-	//check if all selected pods are evicted
+	// check if all selected pods are evicted
 	if err := wait.PollImmediate(time.Second, 30*time.Second, func() (bool, error) {
 		podListEvict, err := clientSet.CoreV1().Pods(rcEvict.Namespace).List(
 			ctx, metav1.ListOptions{LabelSelector: labels.SelectorFromSet(rcEvict.Spec.Template.Labels).String()})
@@ -911,7 +911,8 @@ func TestEvictAnnotation(t *testing.T) {
 			Name: "sample",
 			VolumeSource: v1.VolumeSource{
 				EmptyDir: &v1.EmptyDirVolumeSource{
-					SizeLimit: resource.NewQuantity(int64(10), resource.BinarySI)},
+					SizeLimit: resource.NewQuantity(int64(10), resource.BinarySI),
+				},
 			},
 		},
 	}
@@ -1143,8 +1144,8 @@ func createBalancedPodForNodes(
 
 	// find the max, if the node has the max,use the one, if not,use the ratio parameter
 	var maxCPUFraction, maxMemFraction float64 = ratio, ratio
-	var cpuFractionMap = make(map[string]float64)
-	var memFractionMap = make(map[string]float64)
+	cpuFractionMap := make(map[string]float64)
+	memFractionMap := make(map[string]float64)
 
 	for _, node := range nodes {
 		cpuFraction, memFraction, _, _ := computeCPUMemFraction(t, cs, node, podRequestedResource)
@@ -1181,7 +1182,7 @@ func createBalancedPodForNodes(
 		// add crioMinMemLimit to ensure that all pods are setting at least that much for a limit, while keeping the same ratios
 		needCreateResource[v1.ResourceMemory] = *resource.NewQuantity(int64((ratio-memFraction)*float64(memAllocatableVal)+float64(crioMinMemLimit)), resource.BinarySI)
 
-		var gracePeriod = int64(1)
+		gracePeriod := int64(1)
 		// Don't set OwnerReferences to avoid pod eviction
 		pod := &v1.Pod{
 			ObjectMeta: metav1.ObjectMeta{
@@ -1311,6 +1312,30 @@ func waitForPodRunning(ctx context.Context, t *testing.T, clientSet clientset.In
 	}
 }
 
+func waitForPodsRunning(ctx context.Context, t *testing.T, clientSet clientset.Interface, labelMap map[string]string, desireRunningPodNum int, namespace string) {
+	if err := wait.PollImmediate(10*time.Second, 60*time.Second, func() (bool, error) {
+		podList, err := clientSet.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
+			LabelSelector: labels.SelectorFromSet(labelMap).String(),
+		})
+		if err != nil {
+			return false, err
+		}
+		if len(podList.Items) != desireRunningPodNum {
+			t.Logf("Waiting for %v pods to be running, got %v instead", desireRunningPodNum, len(podList.Items))
+			return false, nil
+		}
+		for _, pod := range podList.Items {
+			if pod.Status.Phase != v1.PodRunning {
+				t.Logf("Pod %v not running yet, is %v instead", pod.Name, pod.Status.Phase)
+				return false, nil
+			}
+		}
+		return true, nil
+	}); err != nil {
+		t.Fatalf("Error waiting for pods running: %v", err)
+	}
+}
+
 func splitNodesAndWorkerNodes(nodes []v1.Node) ([]*v1.Node, []*v1.Node) {
 	var allNodes []*v1.Node
 	var workerNodes []*v1.Node

test/e2e/policy_leaderelection_a.yaml

@@ -0,0 +1,11 @@
+apiVersion: "descheduler/v1alpha1"
+kind: "DeschedulerPolicy"
+strategies:
+  "PodLifeTime":
+    enabled: true
+    params:
+      podLifeTime:
+        maxPodLifeTimeSeconds: 5
+      namespaces:
+        include:
+          - "e2e-testleaderelection-a"

test/e2e/policy_leaderelection_b.yaml

@@ -0,0 +1,11 @@
+apiVersion: "descheduler/v1alpha1"
+kind: "DeschedulerPolicy"
+strategies:
+  "PodLifeTime":
+    enabled: true
+    params:
+      podLifeTime:
+        maxPodLifeTimeSeconds: 5
+      namespaces:
+        include:
+          - "e2e-testleaderelection-b"