Stop modifying ingresses in the Store

In the Ingress Controller we often get objects from our Stores then make modifications. Since these objects are passed by reference or contain references, they are modified in the Store also. Normally this is OK since we don't update the items in the API.

However, now that we update status of Ingresses, we need unmodified versions of the ingress objects.

* DeepCopy when getting ingress objects from the store
* Get fresh ingress from the store before setting status

Author: isaachawley

install/rbac/rbac.yaml

@@ -50,6 +50,7 @@ rules:
   verbs:
   - list
   - watch
+  - get
 - apiGroups:
   - "extensions"
   resources:

nginx-controller/controller/controller.go

@@ -120,31 +120,6 @@ func NewLoadBalancerController(input NewLoadBalancerControllerInput) *LoadBalanc
 
 	glog.V(3).Infof("Nginx Ingress Controller has class: %v", input.IngressClass)
 
-	lbc.statusUpdater = &StatusUpdater{
-		client:              input.KubeClient,
-		namespace:           input.ControllerNamespace,
-		externalServiceName: input.ExternalServiceName,
-	}
-
-	if input.ReportIngressStatus && input.LeaderElectionEnabled {
-		leaderCallbacks := leaderelection.LeaderCallbacks{
-			OnStartedLeading: func(stop <-chan struct{}) {
-				glog.V(3).Info("started leading, updating ingress status")
-				ingresses, mergeableIngresses := lbc.getManagedIngresses()
-				err := lbc.statusUpdater.UpdateManagedAndMergeableIngresses(ingresses, mergeableIngresses)
-				if err != nil {
-					glog.V(3).Infof("error updating status when starting leading: %v", err)
-				}
-			},
-		}
-
-		var err error
-		lbc.leaderElector, err = NewLeaderElector(input.KubeClient, leaderCallbacks, input.ControllerNamespace)
-		if err != nil {
-			glog.V(3).Infof("Error starting LeaderElection: %v", err)
-		}
-	}
-
 	ingHandlers := cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			addIng := obj.(*extensions.Ingress)
@@ -221,6 +196,34 @@ func NewLoadBalancerController(input NewLoadBalancerControllerInput) *LoadBalanc
 		cache.NewListWatchFromClient(lbc.client.Extensions().RESTClient(), "ingresses", input.Namespace, fields.Everything()),
 		&extensions.Ingress{}, input.ResyncPeriod, ingHandlers)
 
+	// statusUpdater requires ingLister to be instantiated, above.
+	lbc.statusUpdater = &StatusUpdater{
+		client:              input.KubeClient,
+		namespace:           input.ControllerNamespace,
+		externalServiceName: input.ExternalServiceName,
+		ingLister:           &lbc.ingLister,
+		keyFunc:             keyFunc,
+	}
+
+	if input.ReportIngressStatus && input.LeaderElectionEnabled {
+		leaderCallbacks := leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(stop <-chan struct{}) {
+				glog.V(3).Info("started leading, updating ingress status")
+				ingresses, mergeableIngresses := lbc.getManagedIngresses()
+				err := lbc.statusUpdater.UpdateManagedAndMergeableIngresses(ingresses, mergeableIngresses)
+				if err != nil {
+					glog.V(3).Infof("error updating status when starting leading: %v", err)
+				}
+			},
+		}
+
+		var err error
+		lbc.leaderElector, err = NewLeaderElector(input.KubeClient, leaderCallbacks, input.ControllerNamespace)
+		if err != nil {
+			glog.V(3).Infof("Error starting LeaderElection: %v", err)
+		}
+	}
+
 	svcHandlers := cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			addSvc := obj.(*api_v1.Service)
@@ -629,7 +632,7 @@ func (lbc *LoadBalancerController) sync(task Task) {
 
 func (lbc *LoadBalancerController) syncIng(task Task) {
 	key := task.Key
-	obj, ingExists, err := lbc.ingLister.Store.GetByKey(key)
+	ing, ingExists, err := lbc.ingLister.GetByKeySafe(key)
 	if err != nil {
 		lbc.syncQueue.requeue(task, err)
 		return
@@ -645,8 +648,6 @@ func (lbc *LoadBalancerController) syncIng(task Task) {
 	} else {
 		glog.V(2).Infof("Adding or Updating Ingress: %v\n", key)
 
-		ing := obj.(*extensions.Ingress)
-
 		if isMaster(ing) {
 			mergeableIngExs, err := lbc.createMergableIngresses(ing)
 			if err != nil {

nginx-controller/controller/status.go

@@ -24,6 +24,8 @@ type StatusUpdater struct {
 	externalStatusAddress    string
 	externalServiceAddresses []string
 	status                   []api_v1.LoadBalancerIngress
+	keyFunc                  func(obj interface{}) (string, error)
+	ingLister                *StoreToIngressLister
 }
 
 // UpdateManagedAndMergeableIngresses handles the full return format of LoadBalancerController.getManagedIngresses
@@ -67,12 +69,27 @@ func (su *StatusUpdater) updateIngressWithStatus(ing v1beta1.Ingress, status []a
 	if reflect.DeepEqual(ing.Status.LoadBalancer.Ingress, status) {
 		return nil
 	}
-	// Objects from the LoadBalancerController.ingLister.Store are retrieved by reference
-	// and are not safe to modify.
-	ingCopy := ing.DeepCopy()
+
+	// Get a pristine Ingress from the Store. Required because annotations can be modified
+	// for mergable Ingress objects and the update status API call will update annotations, not just status.
+	key, err := su.keyFunc(&ing)
+	if err != nil {
+		glog.V(3).Infof("error getting key for ing: %v", err)
+		return err
+	}
+	ingCopy, exists, err := su.ingLister.GetByKeySafe(key)
+	if err != nil {
+		glog.V(3).Infof("error getting ing from Store by key: %v", err)
+		return err
+	}
+	if !exists {
+		glog.V(3).Infof("ing doesn't exist in Store")
+		return nil
+	}
+
 	ingCopy.Status.LoadBalancer.Ingress = status
 	clientIngress := su.client.ExtensionsV1beta1().Ingresses(ingCopy.Namespace)
-	_, err := clientIngress.UpdateStatus(ingCopy)
+	_, err = clientIngress.UpdateStatus(ingCopy)
 	if err != nil {
 		glog.V(3).Infof("error setting ingress status: %v", err)
 		err = su.retryStatusUpdate(clientIngress, ingCopy)

nginx-controller/controller/status_test.go

@@ -0,0 +1,122 @@
+package controller
+
+import (
+	"testing"
+
+	"k8s.io/api/core/v1"
+	extensions "k8s.io/api/extensions/v1beta1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+)
+
+func TestStatusUpdate(t *testing.T) {
+	ing := extensions.Ingress{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Name:      "ing-1",
+			Namespace: "namespace",
+		},
+		Status: extensions.IngressStatus{
+			LoadBalancer: v1.LoadBalancerStatus{
+				Ingress: []v1.LoadBalancerIngress{
+					{
+						IP: "1.2.3.4",
+					},
+				},
+			},
+		},
+	}
+	fakeClient := fake.NewSimpleClientset(
+		&extensions.IngressList{Items: []extensions.Ingress{
+			ing,
+		}},
+	)
+	ingLister := StoreToIngressLister{}
+	ingLister.Store, _ = cache.NewInformer(
+		cache.NewListWatchFromClient(fakeClient.Extensions().RESTClient(), "ingresses", "nginx-ingress", fields.Everything()),
+		&extensions.Ingress{}, 2, nil)
+
+	ingLister.Store.Add(&ing)
+
+	su := StatusUpdater{
+		client:                fakeClient,
+		namespace:             "namespace",
+		externalServiceName:   "service-name",
+		externalStatusAddress: "123.123.123.123",
+		ingLister:             &ingLister,
+		keyFunc:               cache.DeletionHandlingMetaNamespaceKeyFunc,
+	}
+	err := su.ClearIngressStatus(ing)
+	if err != nil {
+		t.Errorf("error clearing ing status: %v", err)
+	}
+	ings, _ := fakeClient.ExtensionsV1beta1().Ingresses("namespace").List(meta_v1.ListOptions{})
+	ingf := ings.Items[0]
+	if !checkStatus("", ingf) {
+		t.Errorf("expected: %v actual: %v", "", ingf.Status.LoadBalancer.Ingress[0])
+	}
+
+	su.SaveStatusFromExternalStatus("1.1.1.1")
+	err = su.UpdateIngressStatus(ing)
+	if err != nil {
+		t.Errorf("error updating ing status: %v", err)
+	}
+	ring, _ := fakeClient.ExtensionsV1beta1().Ingresses(ing.Namespace).Get(ing.Name, meta_v1.GetOptions{})
+	if !checkStatus("1.1.1.1", *ring) {
+		t.Errorf("expected: %v actual: %v", "", ring.Status.LoadBalancer.Ingress)
+	}
+
+	svc := v1.Service{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Namespace: "namespace",
+			Name:      "service-name",
+		},
+		Status: v1.ServiceStatus{
+			LoadBalancer: v1.LoadBalancerStatus{
+				Ingress: []v1.LoadBalancerIngress{v1.LoadBalancerIngress{
+					IP: "2.2.2.2",
+				}},
+			},
+		},
+	}
+	su.SaveStatusFromExternalService(&svc)
+	err = su.UpdateIngressStatus(ing)
+	if err != nil {
+		t.Errorf("error updating ing status: %v", err)
+	}
+	ring, _ = fakeClient.ExtensionsV1beta1().Ingresses(ing.Namespace).Get(ing.Name, meta_v1.GetOptions{})
+	if !checkStatus("1.1.1.1", *ring) {
+		t.Errorf("expected: %v actual: %v", "1.1.1.1", ring.Status.LoadBalancer.Ingress)
+	}
+
+	su.SaveStatusFromExternalStatus("")
+	err = su.UpdateIngressStatus(ing)
+	if err != nil {
+		t.Errorf("error updating ing status: %v", err)
+	}
+	ring, _ = fakeClient.ExtensionsV1beta1().Ingresses(ing.Namespace).Get(ing.Name, meta_v1.GetOptions{})
+	if !checkStatus("2.2.2.2", *ring) {
+		t.Errorf("expected: %v actual: %v", "2.2.2.2", ring.Status.LoadBalancer.Ingress)
+	}
+
+	su.ClearStatusFromExternalService()
+	err = su.UpdateIngressStatus(ing)
+	if err != nil {
+		t.Errorf("error updating ing status: %v", err)
+	}
+	ring, _ = fakeClient.ExtensionsV1beta1().Ingresses(ing.Namespace).Get(ing.Name, meta_v1.GetOptions{})
+	if !checkStatus("", *ring) {
+		t.Errorf("expected: %v actual: %v", "", ring.Status.LoadBalancer.Ingress)
+	}
+}
+
+func checkStatus(expected string, actual extensions.Ingress) bool {
+	if len(actual.Status.LoadBalancer.Ingress) == 0 {
+		if expected == "" {
+			return true
+		}
+		return false
+	}
+	return expected == actual.Status.LoadBalancer.Ingress[0].IP
+}

nginx-controller/controller/utils.go

@@ -163,10 +163,21 @@ type StoreToIngressLister struct {
 	cache.Store
 }
 
+// GetByKeySafe calls Store.GetByKeySafe and returns a copy of the ingress so it is
+// safe to modify.
+func (s *StoreToIngressLister) GetByKeySafe(key string) (ing *extensions.Ingress, exists bool, err error) {
+	item, exists, err := s.Store.GetByKey(key)
+	if !exists || err != nil {
+		return nil, exists, err
+	}
+	ing = item.(*extensions.Ingress).DeepCopy()
+	return
+}
+
 // List lists all Ingress' in the store.
 func (s *StoreToIngressLister) List() (ing extensions.IngressList, err error) {
 	for _, m := range s.Store.List() {
-		ing.Items = append(ing.Items, *(m.(*extensions.Ingress)))
+		ing.Items = append(ing.Items, *(m.(*extensions.Ingress)).DeepCopy())
 	}
 	return ing, nil
 }
@@ -175,7 +186,7 @@ func (s *StoreToIngressLister) List() (ing extensions.IngressList, err error) {
 // Note that this ignores services without the right nodePorts.
 func (s *StoreToIngressLister) GetServiceIngress(svc *api_v1.Service) (ings []extensions.Ingress, err error) {
 	for _, m := range s.Store.List() {
-		ing := *m.(*extensions.Ingress)
+		ing := *m.(*extensions.Ingress).DeepCopy()
 		if ing.Namespace != svc.Namespace {
 			continue
 		}