Add cache comparison for pods and pdbs

pkg/scheduler/factory/BUILD

@@ -112,8 +112,10 @@ go_test(
         "//pkg/scheduler/testing:go_default_library",
         "//pkg/scheduler/util:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/api/policy/v1beta1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/client-go/informers:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",

pkg/scheduler/factory/cache_comparer.go

@@ -22,14 +22,17 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
+	policy "k8s.io/api/policy/v1beta1"
 	"k8s.io/apimachinery/pkg/labels"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	v1beta1 "k8s.io/client-go/listers/policy/v1beta1"
 	"k8s.io/kubernetes/pkg/scheduler/schedulercache"
 )
 
 type cacheComparer struct {
 	nodeLister corelisters.NodeLister
 	podLister  corelisters.PodLister
+	pdbLister  v1beta1.PodDisruptionBudgetLister
 	cache      schedulercache.Cache
 
 	compareStrategy
@@ -44,12 +47,30 @@ func (c *cacheComparer) Compare() error {
 		return err
 	}
 
+	pods, err := c.podLister.List(labels.Everything())
+	if err != nil {
+		return err
+	}
+
+	pdbs, err := c.pdbLister.List(labels.Everything())
+	if err != nil {
+		return err
+	}
+
 	snapshot := c.cache.Snapshot()
 
 	if missed, redundant := c.CompareNodes(nodes, snapshot.Nodes); len(missed)+len(redundant) != 0 {
 		glog.Warningf("cache mismatch: missed nodes: %s; redundant nodes: %s", missed, redundant)
 	}
 
+	if missed, redundant := c.ComparePods(pods, snapshot.Nodes); len(missed)+len(redundant) != 0 {
+		glog.Warningf("cache mismatch: missed pods: %s; redundant pods: %s", missed, redundant)
+	}
+
+	if missed, redundant := c.ComparePdbs(pdbs, snapshot.Pdbs); len(missed)+len(redundant) != 0 {
+		glog.Warningf("cache mismatch: missed pdbs: %s; redundant pdbs: %s", missed, redundant)
+	}
+
 	return nil
 }
 
@@ -57,8 +78,6 @@ type compareStrategy struct {
 }
 
 func (c compareStrategy) CompareNodes(nodes []*v1.Node, nodeinfos map[string]*schedulercache.NodeInfo) (missed, redundant []string) {
-	missed, redundant = []string{}, []string{}
-
 	actual := []string{}
 	for _, node := range nodes {
 		actual = append(actual, node.Name)
@@ -69,6 +88,42 @@ func (c compareStrategy) CompareNodes(nodes []*v1.Node, nodeinfos map[string]*sc
 		cached = append(cached, nodeName)
 	}
 
+	return compareStrings(actual, cached)
+}
+
+func (c compareStrategy) ComparePods(pods []*v1.Pod, nodeinfos map[string]*schedulercache.NodeInfo) (missed, redundant []string) {
+	actual := []string{}
+	for _, pod := range pods {
+		actual = append(actual, string(pod.UID))
+	}
+
+	cached := []string{}
+	for _, nodeinfo := range nodeinfos {
+		for _, pod := range nodeinfo.Pods() {
+			cached = append(cached, string(pod.UID))
+		}
+	}
+
+	return compareStrings(actual, cached)
+}
+
+func (c compareStrategy) ComparePdbs(pdbs []*policy.PodDisruptionBudget, pdbCache map[string]*policy.PodDisruptionBudget) (missed, redundant []string) {
+	actual := []string{}
+	for _, pdb := range pdbs {
+		actual = append(actual, string(pdb.Name))
+	}
+
+	cached := []string{}
+	for pdbName := range pdbCache {
+		cached = append(cached, pdbName)
+	}
+
+	return compareStrings(actual, cached)
+}
+
+func compareStrings(actual, cached []string) (missed, redundant []string) {
+	missed, redundant = []string{}, []string{}
+
 	sort.Strings(actual)
 	sort.Strings(cached)
 

pkg/scheduler/factory/cache_comparer_test.go

@@ -21,6 +21,8 @@ import (
 	"testing"
 
 	"k8s.io/api/core/v1"
+	policy "k8s.io/api/policy/v1beta1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/kubernetes/pkg/scheduler/schedulercache"
 )
 
@@ -77,3 +79,118 @@ func TestCompareNodes(t *testing.T) {
 		}
 	}
 }
+
+func TestComparePods(t *testing.T) {
+	compare := compareStrategy{}
+
+	tests := []struct {
+		actual    []string
+		cached    []string
+		missing   []string
+		redundant []string
+	}{
+		{
+			actual:    []string{"foo", "bar"},
+			cached:    []string{"bar", "foo", "foobar"},
+			missing:   []string{},
+			redundant: []string{"foobar"},
+		},
+		{
+			actual:    []string{"foo", "bar", "foobar"},
+			cached:    []string{"bar", "foo"},
+			missing:   []string{"foobar"},
+			redundant: []string{},
+		},
+		{
+			actual:    []string{"foo", "bar", "foobar"},
+			cached:    []string{"bar", "foobar", "foo"},
+			missing:   []string{},
+			redundant: []string{},
+		},
+	}
+
+	for _, test := range tests {
+		pods := []*v1.Pod{}
+		for _, uid := range test.actual {
+			pod := &v1.Pod{}
+			pod.UID = types.UID(uid)
+			pods = append(pods, pod)
+		}
+
+		nodeInfo := make(map[string]*schedulercache.NodeInfo)
+		for _, uid := range test.cached {
+			pod := &v1.Pod{}
+			pod.UID = types.UID(uid)
+			pod.Namespace = "ns"
+			pod.Name = uid
+
+			nodeInfo[uid] = schedulercache.NewNodeInfo(pod)
+		}
+
+		m, r := compare.ComparePods(pods, nodeInfo)
+
+		if !reflect.DeepEqual(m, test.missing) {
+			t.Errorf("missing expected to be %s; got %s", test.missing, m)
+		}
+
+		if !reflect.DeepEqual(r, test.redundant) {
+			t.Errorf("redundant expected to be %s; got %s", test.redundant, r)
+		}
+	}
+}
+
+func TestComparePdbs(t *testing.T) {
+	compare := compareStrategy{}
+
+	tests := []struct {
+		actual    []string
+		cached    []string
+		missing   []string
+		redundant []string
+	}{
+		{
+			actual:    []string{"foo", "bar"},
+			cached:    []string{"bar", "foo", "foobar"},
+			missing:   []string{},
+			redundant: []string{"foobar"},
+		},
+		{
+			actual:    []string{"foo", "bar", "foobar"},
+			cached:    []string{"bar", "foo"},
+			missing:   []string{"foobar"},
+			redundant: []string{},
+		},
+		{
+			actual:    []string{"foo", "bar", "foobar"},
+			cached:    []string{"bar", "foobar", "foo"},
+			missing:   []string{},
+			redundant: []string{},
+		},
+	}
+
+	for _, test := range tests {
+		pdbs := []*policy.PodDisruptionBudget{}
+		for _, name := range test.actual {
+			pdb := &policy.PodDisruptionBudget{}
+			pdb.Name = name
+			pdbs = append(pdbs, pdb)
+		}
+
+		cache := make(map[string]*policy.PodDisruptionBudget)
+		for _, name := range test.cached {
+			pdb := &policy.PodDisruptionBudget{}
+			pdb.Name = name
+			cache[name] = pdb
+		}
+
+		m, r := compare.ComparePdbs(pdbs, cache)
+
+		if !reflect.DeepEqual(m, test.missing) {
+			t.Errorf("missing expected to be %s; got %s", test.missing, m)
+		}
+
+		if !reflect.DeepEqual(r, test.redundant) {
+			t.Errorf("redundant expected to be %s; got %s", test.redundant, r)
+		}
+	}
+}

pkg/scheduler/factory/factory.go

@@ -301,6 +301,7 @@ func NewConfigFactory(
 	comparer := &cacheComparer{
 		podLister:  podInformer.Lister(),
 		nodeLister: nodeInformer.Lister(),
+		pdbLister:  pdbInformer.Lister(),
 		cache:      c.schedulerCache,
 	}