Deployment: Use ControllerRef to route watch events.

This is part of the completion of ControllerRef, as described here:

https://github.com/kubernetes/community/blob/master/contributors/design-proposals/controller-ref.md#watches

pkg/controller/deployment/BUILD

@@ -33,7 +33,6 @@ go_library(
         "//pkg/util/labels:go_default_library",
         "//pkg/util/metrics:go_default_library",
         "//vendor:github.com/golang/glog",
-        "//vendor:k8s.io/apimachinery/pkg/api/equality",
         "//vendor:k8s.io/apimachinery/pkg/api/errors",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/labels",

pkg/controller/deployment/deployment_controller.go

@@ -23,11 +23,9 @@ package deployment
 import (
 	"fmt"
 	"reflect"
-	"sort"
 	"time"
 
 	"github.com/golang/glog"
-	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -195,17 +193,46 @@ func (dc *DeploymentController) deleteDeployment(obj interface{}) {
 // addReplicaSet enqueues the deployment that manages a ReplicaSet when the ReplicaSet is created.
 func (dc *DeploymentController) addReplicaSet(obj interface{}) {
 	rs := obj.(*extensions.ReplicaSet)
+
+	if rs.DeletionTimestamp != nil {
+		// On a restart of the controller manager, it's possible for an object to
+		// show up in a state that is already pending deletion.
+		dc.deleteReplicaSet(rs)
+		return
+	}
+
+	// If it has a ControllerRef, that's all that matters.
+	if controllerRef := controller.GetControllerOf(rs); controllerRef != nil {
+		if controllerRef.Kind != controllerKind.Kind {
+			// It's controller by a different type of controller.
+			return
+		}
 		glog.V(4).Infof("ReplicaSet %s added.", rs.Name)
-	if d := dc.getDeploymentForReplicaSet(rs); d != nil {
+		d, err := dc.dLister.Deployments(rs.Namespace).Get(controllerRef.Name)
+		if err != nil {
+			return
+		}
+		dc.enqueueDeployment(d)
+		return
+	}
+
+	// Otherwise, it's an orphan. Get a list of all matching Deployments and sync
+	// them to see if anyone wants to adopt it.
+	ds := dc.getDeploymentsForReplicaSet(rs)
+	if len(ds) == 0 {
+		return
+	}
+	glog.V(4).Infof("Orphan ReplicaSet %s added.", rs.Name)
+	for _, d := range ds {
 		dc.enqueueDeployment(d)
 	}
 }
 
-// getDeploymentForReplicaSet returns the deployment managing the given ReplicaSet.
+// getDeploymentsForReplicaSet returns a list of Deployments that potentially
-func (dc *DeploymentController) getDeploymentForReplicaSet(rs *extensions.ReplicaSet) *extensions.Deployment {
+// match a ReplicaSet.
+func (dc *DeploymentController) getDeploymentsForReplicaSet(rs *extensions.ReplicaSet) []*extensions.Deployment {
 	deployments, err := dc.dLister.GetDeploymentsForReplicaSet(rs)
 	if err != nil || len(deployments) == 0 {
-		glog.V(4).Infof("Error: %v. No deployment found for ReplicaSet %v, deployment controller will avoid syncing.", err, rs.Name)
 		return nil
 	}
 	// Because all ReplicaSet's belonging to a deployment should have a unique label key,
@@ -213,11 +240,12 @@ func (dc *DeploymentController) getDeploymentForReplicaSet(rs *extensions.Replic
 	// If that happens we should probably dynamically repair the situation by ultimately
 	// trying to clean up one of the controllers, for now we just return the older one
 	if len(deployments) > 1 {
-		sort.Sort(util.BySelectorLastUpdateTime(deployments))
+		// ControllerRef will ensure we don't do anything crazy, but more than one
+		// item in this list nevertheless constitutes user error.
 		glog.V(4).Infof("user error! more than one deployment is selecting replica set %s/%s with labels: %#v, returning %s/%s",
 			rs.Namespace, rs.Name, rs.Labels, deployments[0].Namespace, deployments[0].Name)
 	}
-	return deployments[0]
+	return deployments
 }
 
 // updateReplicaSet figures out what deployment(s) manage a ReplicaSet when the ReplicaSet
@@ -232,16 +260,45 @@ func (dc *DeploymentController) updateReplicaSet(old, cur interface{}) {
 		// Two different versions of the same replica set will always have different RVs.
 		return
 	}
-	// TODO: Write a unittest for this case
+
-	glog.V(4).Infof("ReplicaSet %s updated.", curRS.Name)
+	curControllerRef := controller.GetControllerOf(curRS)
-	if d := dc.getDeploymentForReplicaSet(curRS); d != nil {
+	oldControllerRef := controller.GetControllerOf(oldRS)
+	controllerRefChanged := !reflect.DeepEqual(curControllerRef, oldControllerRef)
+	if controllerRefChanged &&
+		oldControllerRef != nil && oldControllerRef.Kind == controllerKind.Kind {
+		// The ControllerRef was changed. Sync the old controller, if any.
+		d, err := dc.dLister.Deployments(oldRS.Namespace).Get(oldControllerRef.Name)
+		if err == nil {
 			dc.enqueueDeployment(d)
 		}
-	// A number of things could affect the old deployment: labels changing,
+	}
-	// pod template changing, etc.
+
-	if !apiequality.Semantic.DeepEqual(oldRS, curRS) {
+	// If it has a ControllerRef, that's all that matters.
-		if oldD := dc.getDeploymentForReplicaSet(oldRS); oldD != nil {
+	if curControllerRef != nil {
-			dc.enqueueDeployment(oldD)
+		if curControllerRef.Kind != controllerKind.Kind {
+			// It's controlled by a different type of controller.
+			return
+		}
+		glog.V(4).Infof("ReplicaSet %s updated.", curRS.Name)
+		d, err := dc.dLister.Deployments(curRS.Namespace).Get(curControllerRef.Name)
+		if err != nil {
+			return
+		}
+		dc.enqueueDeployment(d)
+		return
+	}
+
+	// Otherwise, it's an orphan. If anything changed, sync matching controllers
+	// to see if anyone wants to adopt it now.
+	labelChanged := !reflect.DeepEqual(curRS.Labels, oldRS.Labels)
+	if labelChanged || controllerRefChanged {
+		ds := dc.getDeploymentsForReplicaSet(curRS)
+		if len(ds) == 0 {
+			return
+		}
+		glog.V(4).Infof("Orphan ReplicaSet %s updated.", curRS.Name)
+		for _, d := range ds {
+			dc.enqueueDeployment(d)
 		}
 	}
 }
@@ -268,10 +325,23 @@ func (dc *DeploymentController) deleteReplicaSet(obj interface{}) {
 			return
 		}
 	}
-	glog.V(4).Infof("ReplicaSet %s deleted.", rs.Name)
+
-	if d := dc.getDeploymentForReplicaSet(rs); d != nil {
+	controllerRef := controller.GetControllerOf(rs)
-		dc.enqueueDeployment(d)
+	if controllerRef == nil {
+		// No controller should care about orphans being deleted.
+		return
 	}
+	if controllerRef.Kind != controllerKind.Kind {
+		// It's controlled by a different type of controller.
+		return
+	}
+	glog.V(4).Infof("ReplicaSet %s deleted.", rs.Name)
+
+	d, err := dc.dLister.Deployments(rs.Namespace).Get(controllerRef.Name)
+	if err != nil {
+		return
+	}
+	dc.enqueueDeployment(d)
 }
 
 // deletePod will enqueue a Recreate Deployment once all of its pods have stopped running.
@@ -351,11 +421,13 @@ func (dc *DeploymentController) getDeploymentForPod(pod *v1.Pod) *extensions.Dep
 	// Find the owning replica set
 	var rs *extensions.ReplicaSet
 	var err error
-	// Look at the owner reference
+	controllerRef := controller.GetControllerOf(pod)
-	controllerRef := controller.GetControllerOf(&pod.ObjectMeta)
+	if controllerRef == nil {
-	if controllerRef != nil {
+		// No controller owns this Pod.
-		// Not a pod owned by a replica set.
+		return nil
+	}
 	if controllerRef.Kind != extensions.SchemeGroupVersion.WithKind("ReplicaSet").Kind {
+		// Not a pod owned by a replica set.
 		return nil
 	}
 	rs, err = dc.rsLister.ReplicaSets(pod.Namespace).Get(controllerRef.Name)
@@ -363,23 +435,20 @@ func (dc *DeploymentController) getDeploymentForPod(pod *v1.Pod) *extensions.Dep
 		glog.V(4).Infof("Cannot get replicaset %q for pod %q: %v", controllerRef.Name, pod.Name, err)
 		return nil
 	}
-	} else {
+
-		// Fallback to listing replica sets.
+	// Now find the Deployment that owns that ReplicaSet.
-		rss, err := dc.rsLister.GetPodReplicaSets(pod)
+	controllerRef = controller.GetControllerOf(rs)
-		if err != nil {
+	if controllerRef == nil {
-			glog.V(4).Infof("Cannot list replica sets for pod %q: %v", pod.Name, err)
 		return nil
 	}
-		// TODO: Handle multiple replica sets gracefully
+	if controllerRef.Kind != controllerKind.Kind {
-		// For now we return the oldest replica set.
+		return nil
-		if len(rss) > 1 {
-			utilruntime.HandleError(fmt.Errorf("more than one ReplicaSet is selecting pod %q with labels: %+v", pod.Name, pod.Labels))
-			sort.Sort(controller.ReplicaSetsByCreationTimestamp(rss))
 	}
-		rs = rss[0]
+	d, err := dc.dLister.Deployments(rs.Namespace).Get(controllerRef.Name)
+	if err != nil {
+		return nil
 	}
-
+	return d
-	return dc.getDeploymentForReplicaSet(rs)
 }
 
 // worker runs a worker thread that just dequeues items, processes them, and marks them done.

pkg/controller/deployment/deployment_controller_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package deployment
 
 import (
+	"strconv"
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -488,8 +489,6 @@ func TestGetPodMapForReplicaSets(t *testing.T) {
 
 	d := newDeployment("foo", 1, nil, nil, nil, map[string]string{"foo": "bar"})
 
-	// Two ReplicaSets that match labels for both Deployments,
-	// but have ControllerRefs to make ownership explicit.
 	rs1 := newReplicaSet(d, "rs1", 1)
 	rs2 := newReplicaSet(d, "rs2", 1)
 
@@ -545,6 +544,282 @@ func TestGetPodMapForReplicaSets(t *testing.T) {
 	}
 }
 
+func TestAddReplicaSet(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	// Two ReplicaSets that match labels for both Deployments,
+	// but have ControllerRefs to make ownership explicit.
+	rs1 := newReplicaSet(d1, "rs1", 1)
+	rs2 := newReplicaSet(d2, "rs2", 1)
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.objects = append(f.objects, d1, d2, rs1, rs2)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	dc.addReplicaSet(rs1)
+	if got, want := dc.queue.Len(), 1; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+	key, done := dc.queue.Get()
+	if key == nil || done {
+		t.Fatalf("failed to enqueue controller for rs %v", rs1.Name)
+	}
+	expectedKey, _ := controller.KeyFunc(d1)
+	if got, want := key.(string), expectedKey; got != want {
+		t.Errorf("queue.Get() = %v, want %v", got, want)
+	}
+
+	dc.addReplicaSet(rs2)
+	if got, want := dc.queue.Len(), 1; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+	key, done = dc.queue.Get()
+	if key == nil || done {
+		t.Fatalf("failed to enqueue controller for rs %v", rs2.Name)
+	}
+	expectedKey, _ = controller.KeyFunc(d2)
+	if got, want := key.(string), expectedKey; got != want {
+		t.Errorf("queue.Get() = %v, want %v", got, want)
+	}
+}
+
+func TestAddReplicaSetOrphan(t *testing.T) {
+	f := newFixture(t)
+
+	// 2 will match the RS, 1 won't.
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d3 := newDeployment("d3", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d3.Spec.Selector.MatchLabels = map[string]string{"foo": "notbar"}
+
+	// Make the RS an orphan. Expect matching Deployments to be queued.
+	rs := newReplicaSet(d1, "rs1", 1)
+	rs.OwnerReferences = nil
+
+	f.dLister = append(f.dLister, d1, d2, d3)
+	f.objects = append(f.objects, d1, d2, d3)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	dc.addReplicaSet(rs)
+	if got, want := dc.queue.Len(), 2; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+}
+
+func TestUpdateReplicaSet(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	// Two ReplicaSets that match labels for both Deployments,
+	// but have ControllerRefs to make ownership explicit.
+	rs1 := newReplicaSet(d1, "rs1", 1)
+	rs2 := newReplicaSet(d2, "rs2", 1)
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.rsLister = append(f.rsLister, rs1, rs2)
+	f.objects = append(f.objects, d1, d2, rs1, rs2)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	prev := *rs1
+	bumpResourceVersion(rs1)
+	dc.updateReplicaSet(&prev, rs1)
+	if got, want := dc.queue.Len(), 1; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+	key, done := dc.queue.Get()
+	if key == nil || done {
+		t.Fatalf("failed to enqueue controller for rs %v", rs1.Name)
+	}
+	expectedKey, _ := controller.KeyFunc(d1)
+	if got, want := key.(string), expectedKey; got != want {
+		t.Errorf("queue.Get() = %v, want %v", got, want)
+	}
+
+	prev = *rs2
+	bumpResourceVersion(rs2)
+	dc.updateReplicaSet(&prev, rs2)
+	if got, want := dc.queue.Len(), 1; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+	key, done = dc.queue.Get()
+	if key == nil || done {
+		t.Fatalf("failed to enqueue controller for rs %v", rs2.Name)
+	}
+	expectedKey, _ = controller.KeyFunc(d2)
+	if got, want := key.(string), expectedKey; got != want {
+		t.Errorf("queue.Get() = %v, want %v", got, want)
+	}
+}
+
+func TestUpdateReplicaSetOrphanWithNewLabels(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	// RS matches both, but is an orphan.
+	rs := newReplicaSet(d1, "rs1", 1)
+	rs.OwnerReferences = nil
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.rsLister = append(f.rsLister, rs)
+	f.objects = append(f.objects, d1, d2, rs)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	// Change labels and expect all matching controllers to queue.
+	prev := *rs
+	prev.Labels = map[string]string{"foo": "notbar"}
+	bumpResourceVersion(rs)
+	dc.updateReplicaSet(&prev, rs)
+	if got, want := dc.queue.Len(), 2; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+}
+
+func TestUpdateReplicaSetChangeControllerRef(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	rs := newReplicaSet(d1, "rs1", 1)
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.rsLister = append(f.rsLister, rs)
+	f.objects = append(f.objects, d1, d2, rs)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	// Change ControllerRef and expect both old and new to queue.
+	prev := *rs
+	prev.OwnerReferences = []metav1.OwnerReference{*newControllerRef(d2)}
+	bumpResourceVersion(rs)
+	dc.updateReplicaSet(&prev, rs)
+	if got, want := dc.queue.Len(), 2; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+}
+
+func TestUpdateReplicaSetRelease(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	rs := newReplicaSet(d1, "rs1", 1)
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.rsLister = append(f.rsLister, rs)
+	f.objects = append(f.objects, d1, d2, rs)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	// Remove ControllerRef and expect all matching controller to sync orphan.
+	prev := *rs
+	rs.OwnerReferences = nil
+	bumpResourceVersion(rs)
+	dc.updateReplicaSet(&prev, rs)
+	if got, want := dc.queue.Len(), 2; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+}
+
+func TestDeleteReplicaSet(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	// Two ReplicaSets that match labels for both Deployments,
+	// but have ControllerRefs to make ownership explicit.
+	rs1 := newReplicaSet(d1, "rs1", 1)
+	rs2 := newReplicaSet(d2, "rs2", 1)
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.rsLister = append(f.rsLister, rs1, rs2)
+	f.objects = append(f.objects, d1, d2, rs1, rs2)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	dc.deleteReplicaSet(rs1)
+	if got, want := dc.queue.Len(), 1; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+	key, done := dc.queue.Get()
+	if key == nil || done {
+		t.Fatalf("failed to enqueue controller for rs %v", rs1.Name)
+	}
+	expectedKey, _ := controller.KeyFunc(d1)
+	if got, want := key.(string), expectedKey; got != want {
+		t.Errorf("queue.Get() = %v, want %v", got, want)
+	}
+
+	dc.deleteReplicaSet(rs2)
+	if got, want := dc.queue.Len(), 1; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+	key, done = dc.queue.Get()
+	if key == nil || done {
+		t.Fatalf("failed to enqueue controller for rs %v", rs2.Name)
+	}
+	expectedKey, _ = controller.KeyFunc(d2)
+	if got, want := key.(string), expectedKey; got != want {
+		t.Errorf("queue.Get() = %v, want %v", got, want)
+	}
+}
+
+func TestDeleteReplicaSetOrphan(t *testing.T) {
+	f := newFixture(t)
+
+	d1 := newDeployment("d1", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+	d2 := newDeployment("d2", 1, nil, nil, nil, map[string]string{"foo": "bar"})
+
+	// Make the RS an orphan. Expect matching Deployments to be queued.
+	rs := newReplicaSet(d1, "rs1", 1)
+	rs.OwnerReferences = nil
+
+	f.dLister = append(f.dLister, d1, d2)
+	f.rsLister = append(f.rsLister, rs)
+	f.objects = append(f.objects, d1, d2, rs)
+
+	// Create the fixture but don't start it,
+	// so nothing happens in the background.
+	dc, _ := f.newController()
+
+	dc.deleteReplicaSet(rs)
+	if got, want := dc.queue.Len(), 0; got != want {
+		t.Fatalf("queue.Len() = %v, want %v", got, want)
+	}
+}
+
+func bumpResourceVersion(obj metav1.Object) {
+	ver, _ := strconv.ParseInt(obj.GetResourceVersion(), 10, 32)
+	obj.SetResourceVersion(strconv.FormatInt(ver+1, 10))
+}
+
 // generatePodFromRS creates a pod, with the input ReplicaSet's selector and its template
 func generatePodFromRS(rs *extensions.ReplicaSet) *v1.Pod {
 	trueVar := true