Merge pull request #85899 from gongguan/slim_down_lister

slim down some lister expansions

pkg/controller/deployment/deployment_controller.go

@@ -231,7 +231,7 @@ func (dc *DeploymentController) addReplicaSet(obj interface{}) {
 // getDeploymentsForReplicaSet returns a list of Deployments that potentially
 // match a ReplicaSet.
 func (dc *DeploymentController) getDeploymentsForReplicaSet(rs *apps.ReplicaSet) []*apps.Deployment {
-	deployments, err := dc.dLister.GetDeploymentsForReplicaSet(rs)
+	deployments, err := util.GetDeploymentsForReplicaSet(dc.dLister, rs)
 	if err != nil || len(deployments) == 0 {
 		return nil
 	}

pkg/controller/deployment/util/BUILD

@@ -18,11 +18,13 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/api/equality:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/typed/apps/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/listers/apps/v1:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
         "//vendor/k8s.io/utils/integer:go_default_library",
     ],
@@ -46,6 +48,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/names:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/testing:go_default_library",
     ],

pkg/controller/deployment/util/deployment_util.go

@@ -24,18 +24,19 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/klog"
-
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	intstrutil "k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/wait"
 	appsclient "k8s.io/client-go/kubernetes/typed/apps/v1"
+	appslisters "k8s.io/client-go/listers/apps/v1"
+	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/controller"
 	labelsutil "k8s.io/kubernetes/pkg/util/labels"
 	"k8s.io/utils/integer"
@@ -912,3 +913,38 @@ func HasProgressDeadline(d *apps.Deployment) bool {
 func HasRevisionHistoryLimit(d *apps.Deployment) bool {
 	return d.Spec.RevisionHistoryLimit != nil && *d.Spec.RevisionHistoryLimit != math.MaxInt32
 }
+
+// GetDeploymentsForReplicaSet returns a list of Deployments that potentially
+// match a ReplicaSet. Only the one specified in the ReplicaSet's ControllerRef
+// will actually manage it.
+// Returns an error only if no matching Deployments are found.
+func GetDeploymentsForReplicaSet(deploymentLister appslisters.DeploymentLister, rs *apps.ReplicaSet) ([]*apps.Deployment, error) {
+	if len(rs.Labels) == 0 {
+		return nil, fmt.Errorf("no deployments found for ReplicaSet %v because it has no labels", rs.Name)
+	}
+
+	// TODO: MODIFY THIS METHOD so that it checks for the podTemplateSpecHash label
+	dList, err := deploymentLister.Deployments(rs.Namespace).List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+
+	var deployments []*apps.Deployment
+	for _, d := range dList {
+		selector, err := metav1.LabelSelectorAsSelector(d.Spec.Selector)
+		if err != nil {
+			return nil, fmt.Errorf("invalid label selector: %v", err)
+		}
+		// If a deployment with a nil or empty selector creeps in, it should match nothing, not everything.
+		if selector.Empty() || !selector.Matches(labels.Set(rs.Labels)) {
+			continue
+		}
+		deployments = append(deployments, d)
+	}
+
+	if len(deployments) == 0 {
+		return nil, fmt.Errorf("could not find deployments set for ReplicaSet %s in namespace %s with labels: %v", rs.Name, rs.Namespace, rs.Labels)
+	}
+
+	return deployments, nil
+}

pkg/controller/deployment/util/deployment_util_test.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apiserver/pkg/storage/names"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	core "k8s.io/client-go/testing"
 	"k8s.io/kubernetes/pkg/controller"
@@ -1402,3 +1403,84 @@ func TestReplicasAnnotationsNeedUpdate(t *testing.T) {
 		})
 	}
 }
+
+func TestGetDeploymentsForReplicaSet(t *testing.T) {
+	fakeInformerFactory := informers.NewSharedInformerFactory(&fake.Clientset{}, 0*time.Second)
+	var deployments []*apps.Deployment
+	for i := 0; i < 3; i++ {
+		deployment := &apps.Deployment{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("deployment-%d", i),
+				Namespace: "test",
+			},
+			Spec: apps.DeploymentSpec{
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"app": fmt.Sprintf("test-%d", i),
+					},
+				},
+			},
+		}
+		deployments = append(deployments, deployment)
+		fakeInformerFactory.Apps().V1().Deployments().Informer().GetStore().Add(deployment)
+	}
+	var rss []*apps.ReplicaSet
+	for i := 0; i < 5; i++ {
+		rs := &apps.ReplicaSet{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: "test",
+				Name:      fmt.Sprintf("test-replicaSet-%d", i),
+				Labels: map[string]string{
+					"app":   fmt.Sprintf("test-%d", i),
+					"label": fmt.Sprintf("label-%d", i),
+				},
+			},
+		}
+		rss = append(rss, rs)
+	}
+	tests := []struct {
+		name   string
+		rs     *apps.ReplicaSet
+		err    error
+		expect []*apps.Deployment
+	}{
+		{
+			name:   "GetDeploymentsForReplicaSet for rs-0",
+			rs:     rss[0],
+			expect: []*apps.Deployment{deployments[0]},
+		},
+		{
+			name:   "GetDeploymentsForReplicaSet for rs-1",
+			rs:     rss[1],
+			expect: []*apps.Deployment{deployments[1]},
+		},
+		{
+			name:   "GetDeploymentsForReplicaSet for rs-2",
+			rs:     rss[2],
+			expect: []*apps.Deployment{deployments[2]},
+		},
+		{
+			name: "GetDeploymentsForReplicaSet for rs-3",
+			rs:   rss[3],
+			err:  fmt.Errorf("could not find deployments set for ReplicaSet %s in namespace %s with labels: %v", rss[3].Name, rss[3].Namespace, rss[3].Labels),
+		},
+		{
+			name: "GetDeploymentsForReplicaSet for rs-4",
+			rs:   rss[4],
+			err:  fmt.Errorf("could not find deployments set for ReplicaSet %s in namespace %s with labels: %v", rss[4].Name, rss[4].Namespace, rss[4].Labels),
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			get, err := GetDeploymentsForReplicaSet(fakeInformerFactory.Apps().V1().Deployments().Lister(), test.rs)
+			if err != nil {
+				if err.Error() != test.err.Error() {
+					t.Errorf("Error from GetDeploymentsForReplicaSet: %v", err)
+				}
+			} else if !reflect.DeepEqual(get, test.expect) {
+				t.Errorf("Expect deployments %v, but got %v", test.expect, get)
+			}
+		})
+	}
+
+}

pkg/controller/service/BUILD

@@ -12,6 +12,7 @@ go_library(
     deps = [
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",

pkg/controller/service/controller.go

@@ -26,6 +26,7 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -372,7 +373,7 @@ func (s *Controller) syncLoadBalancerIfNeeded(service *v1.Service, key string) (
 }
 
 func (s *Controller) ensureLoadBalancer(service *v1.Service) (*v1.LoadBalancerStatus, error) {
-	nodes, err := s.nodeLister.ListWithPredicate(getNodeConditionPredicate())
+	nodes, err := listWithPredicate(s.nodeLister, getNodeConditionPredicate())
 	if err != nil {
 		return nil, err
 	}
@@ -601,7 +602,7 @@ func nodeSlicesEqualForLB(x, y []*v1.Node) bool {
 	return nodeNames(x).Equal(nodeNames(y))
 }
 
-func getNodeConditionPredicate() corelisters.NodeConditionPredicate {
+func getNodeConditionPredicate() NodeConditionPredicate {
 	return func(node *v1.Node) bool {
 		// We add the master to the node list, but its unschedulable.  So we use this to filter
 		// the master.
@@ -645,7 +646,7 @@ func getNodeConditionPredicate() corelisters.NodeConditionPredicate {
 // nodeSyncLoop handles updating the hosts pointed to by all load
 // balancers whenever the set of nodes in the cluster changes.
 func (s *Controller) nodeSyncLoop() {
-	newHosts, err := s.nodeLister.ListWithPredicate(getNodeConditionPredicate())
+	newHosts, err := listWithPredicate(s.nodeLister, getNodeConditionPredicate())
 	if err != nil {
 		runtime.HandleError(fmt.Errorf("Failed to retrieve current set of nodes from node lister: %v", err))
 		return
@@ -847,3 +848,24 @@ func (s *Controller) patchStatus(service *v1.Service, previousStatus, newStatus
 	_, err := patch(s.kubeClient.CoreV1(), service, updated)
 	return err
 }
+
+// NodeConditionPredicate is a function that indicates whether the given node's conditions meet
+// some set of criteria defined by the function.
+type NodeConditionPredicate func(node *v1.Node) bool
+
+// listWithPredicate gets nodes that matches predicate function.
+func listWithPredicate(nodeLister corelisters.NodeLister, predicate NodeConditionPredicate) ([]*v1.Node, error) {
+	nodes, err := nodeLister.List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+
+	var filtered []*v1.Node
+	for i := range nodes {
+		if predicate(nodes[i]) {
+			filtered = append(filtered, nodes[i])
+		}
+	}
+
+	return filtered, nil
+}

pkg/controller/service/controller_test.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -1428,3 +1429,67 @@ func Test_getNodeConditionPredicate(t *testing.T) {
 		})
 	}
 }
+
+func TestListWithPredicate(t *testing.T) {
+	fakeInformerFactory := informers.NewSharedInformerFactory(&fake.Clientset{}, 0*time.Second)
+	var nodes []*v1.Node
+	for i := 0; i < 5; i++ {
+		var phase v1.NodePhase
+		if i%2 == 0 {
+			phase = v1.NodePending
+		} else {
+			phase = v1.NodeRunning
+		}
+		node := &v1.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("node-%d", i),
+			},
+			Status: v1.NodeStatus{
+				Phase: phase,
+			},
+		}
+		nodes = append(nodes, node)
+		fakeInformerFactory.Core().V1().Nodes().Informer().GetStore().Add(node)
+	}
+
+	tests := []struct {
+		name      string
+		predicate NodeConditionPredicate
+		expect    []*v1.Node
+	}{
+		{
+			name: "ListWithPredicate filter Running node",
+			predicate: func(node *v1.Node) bool {
+				return node.Status.Phase == v1.NodeRunning
+			},
+			expect: []*v1.Node{nodes[1], nodes[3]},
+		},
+		{
+			name: "ListWithPredicate filter Pending node",
+			predicate: func(node *v1.Node) bool {
+				return node.Status.Phase == v1.NodePending
+			},
+			expect: []*v1.Node{nodes[0], nodes[2], nodes[4]},
+		},
+		{
+			name: "ListWithPredicate filter Terminated node",
+			predicate: func(node *v1.Node) bool {
+				return node.Status.Phase == v1.NodeTerminated
+			},
+			expect: nil,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			get, err := listWithPredicate(fakeInformerFactory.Core().V1().Nodes().Lister(), test.predicate)
+			sort.Slice(get, func(i, j int) bool {
+				return get[i].Name < get[j].Name
+			})
+			if err != nil {
+				t.Errorf("Error from ListWithPredicate: %v", err)
+			} else if !reflect.DeepEqual(get, test.expect) {
+				t.Errorf("Expect nodes %v, but got %v", test.expect, get)
+			}
+		})
+	}
+}