Merge pull request kubernetes#5115 from yaroslava-serdiuk/pdb

Add PdbRemainingDisruptions struct

cluster-autoscaler/core/scaledown/pdb/pdb.go

@@ -0,0 +1,82 @@
+/*
+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 pdb
+
+import (
+	"fmt"
+
+	apiv1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/autoscaler/cluster-autoscaler/utils/drain"
+	"k8s.io/klog/v2"
+)
+
+// PdbRemainingDisruptions stores how many discuptiption is left for pdb.
+type PdbRemainingDisruptions struct {
+	pdbs []*policyv1.PodDisruptionBudget
+}
+
+// NewPdbRemainingDisruptions initialize PdbRemainingDisruptions.
+func NewPdbRemainingDisruptions(pdbs []*policyv1.PodDisruptionBudget) *PdbRemainingDisruptions {
+	pdbsCopy := make([]*policyv1.PodDisruptionBudget, len(pdbs))
+	for i, pdb := range pdbs {
+		pdbsCopy[i] = pdb.DeepCopy()
+	}
+	return &PdbRemainingDisruptions{pdbsCopy}
+}
+
+// CanDisrupt return if the pod can be removed.
+func (p *PdbRemainingDisruptions) CanDisrupt(pods []*apiv1.Pod) (bool, *drain.BlockingPod) {
+	for _, pdb := range p.pdbs {
+		selector, err := metav1.LabelSelectorAsSelector(pdb.Spec.Selector)
+		if err != nil {
+			klog.Errorf("Can't get selector for pdb %s", pdb.GetNamespace()+" "+pdb.GetName())
+			return false, nil
+		}
+		count := int32(0)
+		for _, pod := range pods {
+			if pod.Namespace == pdb.Namespace && selector.Matches(labels.Set(pod.Labels)) {
+				count += 1
+				if pdb.Status.DisruptionsAllowed < count {
+					return false, &drain.BlockingPod{Pod: pod, Reason: drain.NotEnoughPdb}
+				}
+			}
+		}
+	}
+	return true, nil
+}
+
+// Update make updates the remaining disruptions for pdb.
+func (p *PdbRemainingDisruptions) Update(pods []*apiv1.Pod) error {
+	for _, pdb := range p.pdbs {
+		selector, err := metav1.LabelSelectorAsSelector(pdb.Spec.Selector)
+		if err != nil {
+			return err
+		}
+		for _, pod := range pods {
+			if pod.Namespace == pdb.Namespace && selector.Matches(labels.Set(pod.Labels)) {
+				if pdb.Status.DisruptionsAllowed < 1 {
+					return fmt.Errorf("Pod can't be removed, pdb is blocking by pdb %s, disruptionsAllowed: %v", pdb.GetNamespace()+"/"+pdb.GetName(), pdb.Status.DisruptionsAllowed)
+				}
+				pdb.Status.DisruptionsAllowed -= 1
+			}
+		}
+	}
+	return nil
+}

cluster-autoscaler/core/scaledown/pdb/pdb_test.go

@@ -0,0 +1,251 @@
+/*
+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 pdb
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	apiv1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	. "k8s.io/autoscaler/cluster-autoscaler/utils/test"
+)
+
+var (
+	one    = intstr.FromInt(1)
+	label1 = "label-1"
+	label2 = "label-2"
+	pdb1   = &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo",
+			Namespace: "ns",
+		},
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			MinAvailable: &one,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					label1: "true",
+				},
+			},
+		},
+		Status: policyv1.PodDisruptionBudgetStatus{
+			DisruptionsAllowed: 1,
+		},
+	}
+	pdb2 = &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "bar",
+			Namespace: "ns",
+		},
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			MinAvailable: &one,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					label2: "true",
+				},
+			},
+		},
+		Status: policyv1.PodDisruptionBudgetStatus{
+			DisruptionsAllowed: 2,
+		},
+	}
+	pdb1Copy = &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo",
+			Namespace: "ns",
+		},
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			MinAvailable: &one,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					label1: "true",
+				},
+			},
+		},
+		Status: policyv1.PodDisruptionBudgetStatus{
+			DisruptionsAllowed: 1,
+		},
+	}
+	pdb2Copy = &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "bar",
+			Namespace: "ns",
+		},
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			MinAvailable: &one,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					label2: "true",
+				},
+			},
+		},
+		Status: policyv1.PodDisruptionBudgetStatus{
+			DisruptionsAllowed: 2,
+		},
+	}
+)
+
+func TestCanDisrupt(t *testing.T) {
+	testCases := []struct {
+		name            string
+		podsLabel1      int
+		podsLabel2      int
+		podsBothLabels  int
+		pdbs            []*policyv1.PodDisruptionBudget
+		pdbsDisruptions [2]int32
+		canDisrupt      bool
+	}{
+		{
+			name:       "No pdbs",
+			podsLabel1: 2,
+			podsLabel2: 1,
+			canDisrupt: true,
+		},
+		{
+			name:            "Not enough pod disruption budgets",
+			podsLabel1:      2,
+			podsLabel2:      1,
+			pdbs:            []*policyv1.PodDisruptionBudget{pdb1, pdb2},
+			pdbsDisruptions: [2]int32{1, 2},
+			canDisrupt:      false,
+		},
+		{
+			name:            "Enough pod disruption budgets",
+			podsLabel1:      2,
+			podsLabel2:      3,
+			pdbs:            []*policyv1.PodDisruptionBudget{pdb1, pdb2},
+			pdbsDisruptions: [2]int32{2, 4},
+			canDisrupt:      true,
+		},
+		{
+			name:            "Pod covered with both PDBs can be moved",
+			podsLabel1:      1,
+			podsLabel2:      1,
+			podsBothLabels:  1,
+			pdbs:            []*policyv1.PodDisruptionBudget{pdb1, pdb2},
+			pdbsDisruptions: [2]int32{1, 1},
+			canDisrupt:      true,
+		},
+		{
+			name:            "Pod covered with both PDBs can't be moved",
+			podsLabel1:      2,
+			podsLabel2:      2,
+			podsBothLabels:  1,
+			pdbs:            []*policyv1.PodDisruptionBudget{pdb1, pdb2},
+			pdbsDisruptions: [2]int32{2, 1},
+			canDisrupt:      false,
+		},
+	}
+	for _, test := range testCases {
+		pdb1.Status.DisruptionsAllowed = test.pdbsDisruptions[0]
+		pdb2.Status.DisruptionsAllowed = test.pdbsDisruptions[1]
+		pdbRemainingDisruptions := NewPdbRemainingDisruptions(test.pdbs)
+		pods := makePodsWithLabel(label1, test.podsLabel1)
+		pods2 := makePodsWithLabel(label2, test.podsLabel2-test.podsBothLabels)
+		if test.podsBothLabels > 0 {
+			addLabelToPods(pods[:test.podsBothLabels], label2)
+		}
+		pods = append(pods, pods2...)
+		got, _ := pdbRemainingDisruptions.CanDisrupt(pods)
+		if got != test.canDisrupt {
+			t.Errorf("%s: CanDisrupt() return %v, want %v", test.name, got, test.canDisrupt)
+		}
+	}
+}
+
+func TestUpdate(t *testing.T) {
+	testCases := []struct {
+		name                   string
+		podsLabel1             int
+		podsLabel2             int
+		podsBothLabels         int
+		pdbs                   []*policyv1.PodDisruptionBudget
+		updatedPdbs            []*policyv1.PodDisruptionBudget
+		pdbsDisruptions        [2]int32
+		updatedPdbsDisruptions [2]int32
+		err                    bool
+	}{
+		{
+			name:                   "Pod covered with both PDBs",
+			podsLabel1:             1,
+			podsLabel2:             1,
+			podsBothLabels:         1,
+			pdbs:                   []*policyv1.PodDisruptionBudget{pdb1, pdb2},
+			updatedPdbs:            []*policyv1.PodDisruptionBudget{pdb1Copy, pdb2Copy},
+			pdbsDisruptions:        [2]int32{1, 1},
+			updatedPdbsDisruptions: [2]int32{0, 0},
+		},
+		{
+			name:           "No PDBs",
+			pdbs:           []*policyv1.PodDisruptionBudget{},
+			updatedPdbs:    []*policyv1.PodDisruptionBudget{},
+			podsLabel1:     2,
+			podsLabel2:     3,
+			podsBothLabels: 1,
+		},
+	}
+	for _, test := range testCases {
+		pdb1.Status.DisruptionsAllowed = test.pdbsDisruptions[0]
+		pdb2.Status.DisruptionsAllowed = test.pdbsDisruptions[1]
+		pdbRemainingDisruptions := NewPdbRemainingDisruptions(test.pdbs)
+		pods := makePodsWithLabel(label1, test.podsLabel1)
+		pods2 := makePodsWithLabel(label2, test.podsLabel2-test.podsBothLabels)
+		if test.podsBothLabels > 0 {
+			addLabelToPods(pods[:test.podsBothLabels], label2)
+		}
+		pods = append(pods, pods2...)
+
+		pdb1Copy.Status.DisruptionsAllowed = test.updatedPdbsDisruptions[0]
+		pdb2Copy.Status.DisruptionsAllowed = test.updatedPdbsDisruptions[1]
+		want := NewPdbRemainingDisruptions(test.updatedPdbs)
+		err := pdbRemainingDisruptions.Update(pods)
+		if err != nil && test.err == false {
+			t.Errorf("%s: Update() return err %v, want nil", test.name, err)
+		}
+		if diff := cmp.Diff(want.pdbs, pdbRemainingDisruptions.pdbs); diff != "" {
+			t.Errorf("Update() diff (-want +got):\n%s", diff)
+		}
+	}
+}
+
+func makePodsWithLabel(label string, amount int) []*apiv1.Pod {
+	pods := []*apiv1.Pod{}
+	for i := 0; i < amount; i++ {
+		pod := &apiv1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            fmt.Sprintf("pod-1-%d", i),
+				Namespace:       "ns",
+				OwnerReferences: GenerateOwnerReferences("rs", "ReplicaSet", "extensions/v1beta1", ""),
+				Labels: map[string]string{
+					label: "true",
+				},
+			},
+			Spec: apiv1.PodSpec{},
+		}
+		pods = append(pods, pod)
+	}
+	return pods
+}
+
+func addLabelToPods(pods []*apiv1.Pod, label string) {
+	for _, pod := range pods {
+		pod.ObjectMeta.Labels[label] = "true"
+	}
+}